:Use Spark to import data

Contexto

Spark es un motor de análisis que procesa grandes volúmenes de datos de manera centralizada. Hologres se integra con Apache Spark y con E-MapReduce (EMR) Spark para ayudarle a construir almacenes de datos de forma eficiente. Hologres proporciona el Spark connector, que le permite escribir datos en Hologres por lotes. Con Spark puede leer datos de varios tipos de orígenes, como archivos, tablas de Hive, tablas de MySQL y tablas de PostgreSQL.

Hologres es compatible con PostgreSQL. Por lo tanto, puede usar Spark para leer datos de Hologres mediante PostgreSQL. Después, extraiga, transforme y cargue (ETL) los datos, y escríbalos de nuevo en Hologres o en otros destinos.

Requisitos previos

• La versión de su instancia de Hologres es V0.9 o posterior. Consulte la versión de su instancia de Hologres en la página de detalles de la instancia dentro de la consola de Hologres. Si la versión de su instancia de Hologres es anterior a V0.9, actualice manualmente su instancia de Hologres en la consola de Hologres o únase al grupo de DingTalk de Hologres para obtener soporte técnico. Para obtener más información sobre cómo actualizar manualmente su instancia de Hologres en la consola de Hologres, consulte Instance upgrades. Para obtener más información sobre cómo obtener soporte técnico, consulte Get online support for Hologres.

• Tiene instalado un entorno de Spark con una versión compatible con Hologres. De este modo, podrá ejecutar el comando spark-shell en dicho entorno.

Uso de conexiones

Cuando usa el Spark connector de Hologres para leer o escribir datos, se utilizan conexiones de Java Database Connectivity (JDBC). Los siguientes factores determinan la cantidad de conexiones necesarias:

• Cantidad de tareas paralelas de Spark: obtenga este valor en la interfaz de usuario de Spark mientras se ejecuta un trabajo.

• Modo de escritura de datos: si usa el Spark connector para escribir datos en el modo fixed copy, cada tarea paralela utiliza una conexión JDBC. Si usa la instrucción INSERT para escribir datos, cada tarea paralela utiliza tantas conexiones JDBC como indique el valor de write_thread_size. Cada tarea paralela de lectura de datos utiliza una conexión JDBC.

• Otros: cuando inicia un trabajo, es posible que se necesite obtener información del esquema. En ese caso, se utiliza una conexión JDBC durante un periodo breve.

Calcule la cantidad total de conexiones JDBC que utiliza un trabajo con las siguientes fórmulas:

• Modo fixed copy: Parallelism × 1 + 1

• Instrucción INSERT: Parallelism × write_thread_size + 1

Escritura de datos en Hologres con el Spark connector (recomendado)

Se recomienda usar el Spark connector integrado de Hologres para escribir datos en Hologres. El Spark connector se utiliza junto con Holo Client. En comparación con otros métodos de escritura, el Spark connector ofrece un mejor rendimiento. Siga estos pasos para escribir datos con el Spark connector. Para obtener más información sobre el código de ejemplo, consulte la sección Example of using the Spark connector to write data to Hologres de este tema.

Lectura de datos de un origen específico con Spark y escritura en Hologres

1. Lea los datos de un origen específico.

   Con Spark puede leer datos de distintos tipos de orígenes. A continuación se muestra cómo leer datos de Hologres o de otro tipo de origen:

   • Leer datos de Hologres

     Hologres es compatible con PostgreSQL. Por ello, puede usar Spark para leer datos de Hologres mediante el controlador JDBC de PostgreSQL. El siguiente código de ejemplo sirve únicamente como referencia.

     Nota

     Antes de leer datos de Hologres, descargue el paquete PostgreSQL JDBC JAR desde el sitio web oficial. En este ejemplo se utiliza postgresql-42.2.18. Luego, ejecute el comando ./bin/spark-shell --jars /path/to/postgresql-42.2.18.jar en spark-shell para cargar el paquete JAR de JDBC de PostgreSQL. También puede cargar el paquete JAR de JDBC de PostgreSQL junto con el paquete JAR del Spark connector de Hologres.

     // Read from some table, for example: tb008
     val readDf = spark.read
       .format("jdbc") // Read Hologres data based on the PostgreSQL JDBC driver.
       .option("driver","org.postgresql.Driver")
       .option("url", "jdbc:postgresql://Ip:Por/test_database")
       .option("dbtable", "tb008")
       .option("user", "your_username")
       .option("password", "your_password")
       .load()

     Las versiones V1.3.2 y posteriores del Spark connector permiten leer datos de Hologres y optimizan el rendimiento de las lecturas paralelas. A diferencia del controlador JDBC de PostgreSQL, el Spark connector le permite configurar el paralelismo de lectura y admite la fragmentación de datos en Hologres para implementar lecturas paralelas. Esto mejora significativamente el rendimiento. Código de ejemplo:

     val spark = SparkSession
     .builder
     .appName("ReadFromHologres")
     .master("local[*]")
     .getOrCreate()
     
     spark.sparkContext.setLogLevel("WARN")
     
     import spark.implicits._
     
     val schema = StructType(Array(
       StructField("id", LongType),
       StructField("counts", IntegerType),
       StructField("name", StringType, false),
       StructField("price", DecimalType(38, 12)),
       StructField("out_of_stock", BooleanType)
     ))
     
     val readDf = spark.read
     .format("hologres")
     .schema(schema) // Optional. If you do not specify a schema, all fields in the Hologres table are read by default.
     .option("username", "your_username")
     .option("password", "your_password")
     .option("jdbcurl", "jdbc:postgresql://hologres_endpoint/test_db")
     .option("table", "tb008")
     .option("scan_parallelism", "10") // The default parallelism for reading data from Hologres. The maximum value is the shard count of the Hologres table.
     .load()

   • Leer datos de otro tipo de origen, como un archivo Parquet

     Con Spark también puede leer datos de otros tipos de orígenes, como un archivo Parquet o una tabla de Hive. El siguiente ejemplo proporciona el código correspondiente:

     import org.apache.spark.{SparkConf, SparkContext}
     import org.apache.spark.sql.hive.HiveContext
     
     val sparkConf = new SparkConf()
     val sc = new SparkContext(sparkConf)
     val hiveContext = new HiveContext(sc)
     
     // Read from some table, for example: phone
     val readDf = hiveContext.sql("select * from hive_database.phone")

2. Escriba los datos en Hologres.

   import com.alibaba.hologres.spark2.sink.SourceProvider
   
   -- Write to hologres table
   df.write
     .format("hologres")
     .option(SourceProvider.USERNAME, "your_username")
     .option(SourceProvider.PASSWORD, "your_password")
     .option(SourceProvider.ENDPOINT, "Ip:Port")
     .option(SourceProvider.DATABASE, "test_database")
     .option(SourceProvider.TABLE, table)
     .option(SourceProvider.WRITE_BATCH_SIZE, 512) -- The maximum number of requests allowed in a batch.
     .option(SourceProvider.INPUT_DATA_SCHEMA_DDL, df.schema.toDDL) -- This parameter is required only for Spark 3.X.
     .mode(SaveMode.Append) // This parameter is required only for Spark 3.X.
     .save()

Escritura de datos en Hologres en tiempo real con Spark

1. Ejecute la siguiente instrucción para crear la tabla donde desea escribir los datos en Hologres:

   CREATE TABLE test_table_stream
   (
       value text,
       count bigint
   );

2. Lea los datos de su máquina local. Recopile estadísticas de frecuencia de palabras y escríbalas en Hologres en tiempo real. El siguiente ejemplo proporciona el código correspondiente:

   • Código

      val spark = SparkSession
           .builder
           .appName("StreamToHologres")
           .master("local[*]")
           .getOrCreate()
     
         spark.sparkContext.setLogLevel("WARN")
         import spark.implicits._
     
         val lines = spark.readStream
           .format("socket")
           .option("host", "localhost")
           .option("port", 9999)
           .load()
     
         -- Split the lines into words
         val words = lines.as[String].flatMap(_.split(" "))
     
         -- Generate running word count
         val wordCounts = words.groupBy("value").count()
     
         wordCounts.writeStream
             .outputMode(OutputMode.Complete())
             .format("hologres")
             .option(SourceProvider.USERNAME, "your_username")
             .option(SourceProvider.PASSWORD, "your_password")
             .option(SourceProvider.JDBCURL, "jdbc:postgresql://Ip:Port/dbname")
             .option(SourceProvider.TABLE, "test_table_stream")
             .option("batchsize", 1)
             .option("isolationLevel", "NONE")
             .option("checkpointLocation", checkpointLocation)
             .start()
             .awaitTermination()

   • Parámetros

     Parámetro	Valor predeterminado	Obligatorio	Descripción
     username	Ninguna	Sí	El AccessKey ID de su cuenta de Alibaba Cloud. Obtenga el AccessKey ID en la página Security Management. Se recomienda configurar variables de entorno y obtener el AccessKey ID y el AccessKey secret a partir de ellas. Esto ayuda a reducir el riesgo de filtraciones.
     password	Ninguna	Sí	El AccessKey secret de su cuenta de Alibaba Cloud. Obtenga el AccessKey secret en la página Security Management. Se recomienda configurar variables de entorno y obtener el AccessKey ID y el AccessKey secret a partir de ellas. Esto ayuda a reducir el riesgo de filtraciones.
     table	Ninguna	Sí	El nombre de la tabla de Hologres donde desea escribir los datos.
     endpoint	Ninguna	Configure este parámetro o el parámetro JDBCURL.	El endpoint de su instancia de Hologres. Obtenga el endpoint de su instancia de Hologres en la sección Network Information de la página de detalles de la instancia en la consola de Hologres.
     database	Ninguna	Configure este parámetro o el parámetro JDBCURL.	El nombre de la base de datos de Hologres donde reside la tabla de destino.
     jdbcurl	Ninguna	Configure solo este parámetro o bien los parámetros ENDPOINT y DATABASE.	La JDBC URL de su instancia de Hologres.
     copy_write_mode	true	No	Indica si se escriben los datos en el modo fixed copy. El modo fixed copy es una nueva función admitida en Hologres V1.3. En este modo, los datos se escriben mediante streaming en lugar de por lotes. Por lo tanto, la escritura en modo fixed copy ofrece mayor rendimiento y menor latencia de datos, además de consumir menos recursos de memoria del cliente que la escritura mediante la instrucción INSERT. Nota Si desea usar el modo fixed copy, la versión del conector debe ser V1.3.0 o posterior, y la versión del motor de Hologres debe ser V1.3.34 o posterior.
     copy_write_format	false	No	Indica si se comprueban los datos erróneos. Este parámetro solo surte efecto cuando COPY_WRITE_MODE se establece en true. Si lo establece en true y se generan datos erróneos, el sistema puede identificar la fila cuya escritura falló. Nota La comprobación de datos erróneos afecta negativamente el rendimiento de escritura. Se recomienda establecer este parámetro en true solo durante el proceso de solución de problemas.
     bulk_load	true	No	Indica si se escriben los datos en Hologres en el modo batch copy. En el modo fixed copy, los datos se escriben mediante streaming. Nota En Hologres V2.1 se optimizó el rendimiento de escritura en tablas sin claves primarias. En esta versión, la escritura por lotes en una tabla sin clave primaria no adquiere un bloqueo de tabla, sino un bloqueo de fila. Así, las escrituras por lotes pueden ejecutarse en paralelo con las operaciones realizadas mediante fixed plans. Esto mejora la eficiencia y el paralelismo del procesamiento de datos. Si la versión del conector es V1.4.0 o posterior, la versión del motor de Hologres debe ser V2.1.0 o posterior.
     max_cell_buffer_size	20971520 (20 MB)	No	La longitud máxima de un campo cuando COPY_WRITE_MODE se establece en true.
     copy_write_dirty_data_check	false	No	Indica si se comprueban los datos erróneos. Si establece este parámetro en true y se generan datos erróneos, el sistema puede identificar la fila cuya escritura falló. La comprobación de datos erróneos afecta negativamente el rendimiento de escritura. Se recomienda establecer este parámetro en true solo durante el proceso de solución de problemas. Nota Este parámetro solo surte efecto cuando COPY_WRITE_MODE se establece en true.
     copy_write_direct_connect	true para escenarios que permiten conexiones directas	No	Este parámetro solo surte efecto cuando COPY_WRITE_MODE se establece en true. La cantidad de datos que se pueden escribir en modo copy depende del rendimiento del endpoint de VPC. El sistema comprueba si el entorno puede conectarse directamente al nodo FE de Hologres al escribir datos en modo copy. Si el entorno admite la conexión directa al nodo FE de Hologres, se utilizan conexiones directas de forma predeterminada. Si establece este parámetro en false, no se utilizan conexiones directas.
     input_data_schema_ddl	Ninguna	Este parámetro es obligatorio para Spark 3.X. Configure el valor del parámetro con el formato <your_DataFrame>.schema.toDDL.	La instrucción DDL para DataFrame en Spark.
     write_mode	INSERT_OR_REPLACE	No	La política utilizada para gestionar conflictos de claves primarias. Este parámetro es obligatorio si la tabla de destino tiene una clave primaria. Valores válidos: INSERT_OR_IGNORE: descarta los datos que intenta escribir si ocurre un conflicto de clave primaria. INSERT_OR_UPDATE: actualiza las columnas correspondientes en la tabla de destino si ocurre un conflicto de clave primaria. INSERT_OR_REPLACE: actualiza todas las columnas de la tabla de destino si ocurre un conflicto de clave primaria.
     write_batch_size	512	No	La cantidad máxima de solicitudes permitidas en un lote dentro de un hilo para escribir datos. Si la cantidad total de solicitudes PUT alcanza el límite superior especificado por el parámetro write_batch_size después de gestionar los conflictos según el parámetro write_mode, los datos se envían en un lote.
     write_batch_byte_size	2 MB	No	La cantidad máxima de bytes permitidos en un lote dentro de un hilo para escribir datos. Si el total de bytes de las solicitudes PUT alcanza el límite superior especificado por el parámetro WRITE_BATCH_BYTE_SIZE después de gestionar los conflictos según el parámetro WRITE_MODE, los datos se envían en un lote.
     write_max_interval_ms	10000 ms	No	El intervalo con el que se envían los datos en un lote.
     write_fail_strategy	TYR_ONE_BY_ONE	No	La política utilizada para gestionar errores de envío. Si no se puede enviar un lote, Holo Client envía las entradas de datos del lote una por una en el orden especificado. Si una entrada de datos no se puede enviar, Holo Client devuelve un mensaje de error que contiene la información de dicha entrada.
     write_thread_size	1	No	La cantidad de hilos paralelos utilizados para escribir datos. Cada hilo ocupa una conexión. La cantidad total de conexiones que ocupa un trabajo de Spark varía según el paralelismo de Spark. Calcule la cantidad total de conexiones con la siguiente fórmula: Total number of connections = Value of spark.default.parallelism × Value of WRITE_THREAD_SIZE.
     dynamic_partition	false	No	Indica si se crea automáticamente una partición cuando los datos se escriben en una tabla principal sin particiones. Valores válidos: true y false. true: se crea automáticamente una partición cuando los datos se escriben en una tabla principal sin particiones.
     retry_count	3	No	La cantidad máxima de reintentos permitidos para escribir y consultar datos si ocurre un error de conexión.
     retry_sleep_init_ms	1000 ms	No	El tiempo que consumen los reintentos de una solicitud se calcula con la siguiente fórmula: retry_sleep_init_ms + retry_count × retry_sleep_step_ms.
     retry_sleep_step_ms	10*1000 ms	No	El tiempo que consumen los reintentos de una solicitud se calcula con la siguiente fórmula: retry_sleep_init_ms + retry_count × retry_sleep_step_ms.
     connection_max_idle_ms	60000 ms	No	El tiempo de espera de inactividad para las conexiones utilizadas en la lectura y escritura de datos. Si una conexión permanece inactiva durante un periodo mayor al tiempo de espera especificado, Holo Client libera automáticamente la conexión.
     fixed_connection_mode	false	No	Indica si se usan conexiones fijas. En un modo distinto a fixed copy, como el modo INSERT, las escrituras de datos y las consultas puntuales no ocupan conexiones. Nota La función de conexión fija se encuentra en versión beta y solo está disponible cuando la versión del conector es V1.2.0 o posterior y la versión del motor de Hologres es V1.3.0 o posterior.
     scan_batch_size	256	No	La cantidad de filas que se analizan cada vez que lee datos de Hologres.
     scan_timeout_seconds	60	No	El tiempo de espera de la operación de análisis cuando lee datos de Hologres. Unidad: segundos.
     scan_parallelism	10	No	La cantidad de fragmentos utilizados para leer datos de Hologres. El valor máximo corresponde a la cantidad de fragmentos de la tabla de Hologres. Durante la ejecución del trabajo, los fragmentos se asignan a las tareas de Spark para la lectura de datos.

Asignaciones de tipos de datos

La siguiente tabla describe las asignaciones de tipos de datos entre Spark y Hologres.