Realtime Compute for Apache Flink:Trabajos con dialecto Hive

Requisitos previos

• Si accede a la consola como usuario de RAM, rol de RAM u otra identidad, asegúrese de contar con los permisos obligatorios. Para obtener más información, consulte permissions.

• Debe haber creado un espacio de trabajo. Para obtener más información, consulte Activate Realtime Compute for Apache Flink.

Limitaciones

• Solo Ververica Runtime (VVR) 8.0.11 y versiones posteriores admiten el dialecto Hive.

• Actualmente, los trabajos SQL solo admiten la sintaxis INSERT Statements del dialecto Hive. Debe declarar USE Catalog <yourHiveCatalog> antes de las instrucciones INSERT Statements. Si necesita crear una tabla, realice la operación en la página Scripts.

• No se admiten las funciones definidas por el usuario (UDF) de Hive ni de Flink.

Paso 1: Crear un catálogo de Hive

1. Configure los metadatos de Hive. Para obtener más información, consulte Configure Hive metadata.

2. Cree un catálogo de Hive. Para obtener más información, consulte Create a Hive catalog.

   En este tutorial, el catálogo de Hive se denomina hdfshive.

Paso 2: Preparar tablas de Hive de ejemplo

1. En el panel de navegación izquierdo, vaya a Development > Scripts. Haga clic en New para crear un script.

2. Ejecute las siguientes instrucciones SQL de ejemplo.

   Importante

   La tabla de origen y la tabla de destino de Hive deben ser tablas permanentes creadas con la instrucción CREATE TABLE. No es posible utilizar tablas temporales creadas con la instrucción CREATE TEMPORARY TABLE.

   -- Use the Hive catalog. In this example, the catalog is named hdfshive and was created in Step 1.
   USE CATALOG hdfshive;   
   -- Create a source table with the default storage format.
   CREATE TABLE source_table (
    id INT,
    name STRING,
    age INT,
    city STRING,
    salary FLOAT
   )WITH ('connector' = 'hive');
   -- Create a sink table with the default storage format.
   CREATE TABLE target_table (
   city STRING,
   avg_salary FLOAT,
   user_count INT
   )WITH ('connector' = 'hive');
   -- Insert sample data into the source table.
   INSERT INTO source_table VALUES
   (1, 'Alice', 25, 'New York', 5000.0),
   (2, 'Bob', 30, 'San Francisco', 6000.0),
   (3, 'Charlie', 35, 'New York', 7000.0),
   (4, 'David', 40, 'San Francisco', 8000.0),
   (5, 'Eva', 45, 'Los Angeles', 9000.0);
   -- Create a table with a specific storage format, for example, Parquet.
   -- Load the Hive module.
   load MODULE hive with ('hive-version' = '2.3.6');
   use CATALOG `hdfshive`;
   -- Required: Set the SQL dialect to 'hive' to recognize Hive DDL keywords such as 'STORED'.
   set 'table.sql-dialect' = 'hive';
   CREATE TABLE `parquet_table`(
    id INT,
    name STRING,
    age INT,
    city STRING,
    salary FLOAT
   )STORED AS PARQUET;

Paso 3: Crear un trabajo de Hive SQL

1. En el panel de navegación izquierdo, vaya a Development > ETL.

2. Haga clic en New. En el cuadro de diálogo New Draft, seleccione Blank Batch Draft (BETA) y haga clic en Next.

3. Introduzca la información del trabajo.

   Parámetro	Descripción	Ejemplo
   Name	Nombre del trabajo. Nota El nombre del trabajo debe ser único dentro del espacio de trabajo actual.	hive-sql
   Location	Carpeta donde se almacena el archivo de código del trabajo. También puede hacer clic en el icono situado a la derecha de una carpeta existente para crear una subcarpeta.	Drafts
   Engine version	Versión del motor Flink que utiliza el trabajo. Se recomienda seleccionar una versión con la etiqueta RECOMMENDED. Estas versiones ofrecen mayor confiabilidad y rendimiento. Para obtener más información sobre las versiones del motor, consulte Release notes y Engine versions.	vvr-8.0.11-flink-1.17
   SQL dialect	Lenguaje SQL para el procesamiento de datos. Nota Este parámetro solo aparece si selecciona una versión del motor compatible con el dialecto Hive.	Hive SQL

4. Haga clic en Create.

Paso 4: Escribir y desplegar el trabajo de Hive SQL

1. Escriba las instrucciones SQL.

   Este ejemplo calcula la cantidad de usuarios mayores de 30 años y el salario promedio por ciudad. Puede copiar el siguiente script SQL en el editor SQL.

   -- Use the Hive catalog. In this example, the catalog is named hdfshive and was created in Step 1.
   USE CATALOG hdfshive; 
   INSERT INTO TABLE target_table
   SELECT
     city,
     AVG(salary) AS avg_salary, -- Calculate the average salary
     COUNT(id) AS user_count -- Count the number of users
   FROM source_table
   WHERE age > 30 -- Filter for users older than 30
   GROUP BY city; -- Group by city

2. En la esquina superior derecha, haga clic en Deploy. En el cuadro de diálogo, configure los parámetros según sea necesario (este tutorial utiliza la configuración predeterminada) y haga clic en OK.

(Opcional) Paso 5: Configurar parámetros de tiempo de ejecución

1. En el panel de navegación izquierdo, vaya a O&M > Deployments.

2. En la lista desplegable, seleccione BATCH. Busque el trabajo deseado y haga clic en Details en la columna Actions.

3. En el panel de detalles del despliegue, haga clic en Edit en la sección Runtime parameters configuration.

4. En el campo Other Configuration, agregue la siguiente configuración:

   fs.oss.jindo.endpoint: <YOUR_Endpoint> 
   fs.oss.jindo.buckets: <YOUR_Buckets>
   fs.oss.jindo.accessKeyId: <YOUR_AccessKeyId>
   fs.oss.jindo.accessKeySecret: <YOUR_AccessKeySecret>

   Para obtener más información sobre estos parámetros, consulte Write data to OSS-HDFS.

5. Haga clic en Save.

Paso 6: Iniciar el trabajo y consultar los resultados

1. En la página Deployments, seleccione Batch job en el filtro, busque el trabajo deseado (por ejemplo, hive-sql) y haga clic en Start en la columna Actions.

2. Una vez que el estado del trabajo cambie a FINISHED, consulte los resultados.

   En la página Development > Scripts, ejecute la siguiente instrucción SQL para ver los datos, que incluyen la cantidad de usuarios mayores de 30 años y su salario promedio en cada ciudad.

   -- Use the Hive catalog. In this example, the catalog is named hdfshive and was created in Step 1.
   USE CATALOG hdfshive; 
   select * from target_table;

   La consulta devuelve tres filas de target_table con las columnas city, avg_salary y user_count: Los Ángeles (9000.0, 1), Nueva York (7000.0, 1) y San Francisco (8000.0, 1).

Desarrollo de trabajos JAR de Hive

Es posible ejecutar trabajos con dialecto Hive como trabajos JAR. Esto requiere la versión 11.2 o posterior del paquete JAR "ververica-connector-hive-2.3.6". También debe asegurarse de que las configuraciones de Hive en su trabajo JAR coincidan con la configuración de la consola.

1. Configuración de la consola

   1. El URI del JAR especifica el paquete JAR cargado para el trabajo JAR.

   2. En Additional Dependencies, cargue los cuatro archivos de configuración de su clúster de Hive: core-site.xml, mapred-site.xml, hdfs-site.xml y hive-site.xml. También debe cargar el paquete JAR ververica-connector-hive-2.3.6.

   3. Configure los parámetros de tiempo de ejecución. Según la configuración de su clúster de Hive, si necesita escribir datos en OSS-HDFS, utilice la configuración de (Optional) Step 5: Configure runtime parameters.

      table.sql-dialect: HIVE
      classloader.parent-first-patterns.additional: org.apache.hadoop;org.antlr.runtime
      kubernetes.application-mode.classpath.include-user-jar: true

2. Código de ejemplo de trabajo JAR:

   1. StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
      Configuration conf = new Configuration();
      conf.setString("type", "hive");
      conf.setString("default-database", "default");
      conf.setString("hive-version", "2.3.6"); 
      conf.setString("hive-conf-dir", "/flink/usrlib/" );
      conf.setString("hadoop-conf-dir", "/flink/usrlib/");
      CatalogDescriptor descriptor = CatalogDescriptor.of("hivecat", conf);
      tableEnv.createCatalog("hivecat", descriptor);
      tableEnv.loadModule("hive", new HiveModule());
      tableEnv.useModules("hive");
      tableEnv.useCatalog("hivecat");
      tableEnv.executeSql("insert into `hivecat`.`default`.`test_write` select * from `hivecat`.`default`.`test_read`;");

Documentos relacionados

• Para obtener más información sobre la sintaxis de la instrucción INSERT para el dialecto Hive, consulte INSERT Statements | Apache Flink.

• Para obtener más información sobre el procesamiento de datos por lotes con Flink SQL, consulte Batch processing.