Hologres:Use Spark to write data to Hologres

Prerequisites

• The version of your Hologres instance is V0.9 or later. You can view the version of your Hologres instance on the instance details page in the Hologres console. If the version of your Hologres instance is earlier than V0.9, manually upgrade your Hologres instance in the Hologres console or join the Hologres DingTalk group for technical support. For more information about how to manually upgrade your Hologres instance in the Hologres console, see Instance upgrades. For more information about how to obtain technical support, see Obtain online support for Hologres.

• A Spark environment that is of a version supported by Hologres is installed. This way, you can run the spark-shell command in the Spark environment.

(Recommended) Use the Spark connector to write data to Hologres

We recommend that you use the built-in Spark connector of Hologres to write data to Hologres. The Spark connector is used together with Holo Client. Compared with other methods of writing data, the Spark connector provides better write performance. To use the Spark connector to write data, perform the following steps. For more information about the sample code, see the Example of using the Spark connector to write data to Hologres section in this topic.

Use Spark to read data from a specific type of source and write the data to Hologres

1. Read data from a specific type of source.

   You can use Spark to read data from different types of sources. The following examples show how to read data from Hologres or another type of source:

   • Read data from Hologres

     Hologres is compatible with PostgreSQL. You can use Spark to read Hologres data based on the PostgreSQL JDBC driver. The following sample code is for reference only.

     Note

     Before you read Hologres data, download the PostgreSQL JDBC JAR package on the official website. In this example, postgresql-42.2.18 is used. Then, run the ./bin/spark-shell --jars /path/to/postgresql-42.2.18.jar command in spark-shell to load the PostgreSQL JDBC JAR package. You can also load the PostgreSQL JDBC JAR package together with the JAR package of the Spark connector.

     // 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()

   • Read data from another type of source such as a Parquet file

     You can use Spark to read data from other types of sources, such as a Parquet file or a Hive table. The following example provides the sample code:

     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. Write the data to 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) -- The DDL statement for DataFrame. This parameter is required only for Spark 3.X.
     .mode(SaveMode.Append) // The save mode. This parameter is required only for Spark 3.X.
     .save()

Use Spark to write data to Hologres in real time

1. Execute the following statement to create a table to which you want to write data in Hologres:

   CREATE TABLE test_table_stream
   (
       value text,
       count bigint
   );

2. Read data from your on-premises machine. Collect word frequency statistics and write the statistics to Hologres in real time. The following example provides sample code:

   • Code

      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()

   • Parameters

     Parameter	Default value	Required	Description
     USERNAME	No default value	Yes	The AccessKey ID of your Alibaba Cloud account. You can obtain the AccessKey ID on the Security Management page. We recommend that you configure environment variables and obtain the AccessKey ID and AccessKey secret from the environment variables. This helps reduce the leak risk.
     PASSWORD	No default value	Yes	The AccessKey secret of your Alibaba Cloud account. You can obtain the AccessKey secret on the Security Management page. We recommend that you configure environment variables and obtain the AccessKey ID and AccessKey secret from the environment variables. This helps reduce the leak risk.
     TABLE	No default value	Yes	The name of the Hologres table to which you want to write data.
     ENDPOINT	No default value	Configure this parameter or the JDBCURL parameter.	The endpoint of your Hologres instance. You can obtain the endpoint of your Hologres instance in the Network Information section of the instance details page in the Hologres console.
     DATABASE	No default value	Configure this parameter or the JDBCURL parameter.	The name of the Hologres database in which the destination table resides.
     JDBCURL	No default value	Configure only this parameter or the ENDPOINT and DATABASE parameters.	The JDBC URL of your Hologres instance.
     COPY_WRITE_MODE	true	No	Specifies whether to write data in the fixed copy mode. Fixed copy is a new feature that is supported in Hologres V1.3. In the fixed copy mode, data is written in the streaming mode rather than in batches. Therefore, data writing in the fixed copy mode provides higher throughput and lower data latency, and consumes less client memory resources than data writing by using the INSERT statement. Note If you want to use the fixed copy mode, the connector version must be V1.3.0 or later, and the Hologres engine version must be r1.3.34 or later.
     COPY_WRITE_DIRTY_DATA_CHECK	false	No	Specifies whether to check dirty data. This parameter takes effect only when COPY_WRITE_MODE is set to true. If you set this parameter to true and dirty data is generated, the system can locate the row to which data fails to be written. Note Checking of dirty data negatively affects the write performance. We recommend that you set this parameter to true only in the troubleshooting process.
     COPY_WRITE_DIRECT_CONNECT	true for scenarios that allow direct connections	No	This parameter takes effect only when COPY_WRITE_MODE is set to true. The amount of data that can be written in the copy mode is determined based on the throughput of the VPC endpoint. The system checks whether the environment can directly connect to the Hologres FE node when data is written in the copy mode. Direct connections are used by default if the environment can directly connect to the Hologres FE node. If this parameter is set to false, direct connections are not used.
     INPUT_DATA_SCHEMA_DDL	No default value	This parameter is required for Spark 3.X. Set this parameter to <your_DataFrame>.schema.toDDL.	The DDL statement for DataFrame in Spark.
     WRITE_MODE	INSERT_OR_REPLACE	No	The policy that is used to handle primary key conflicts. This parameter is required if the destination table has a primary key. Valid values: INSERT_OR_IGNORE: discards the data that you want to write if a primary key conflict occurs. INSERT_OR_UPDATE: updates the relevant columns in the destination table if a primary key conflict occurs. INSERT_OR_REPLACE: updates all columns in the destination table if a primary key conflict occurs.
     WRITE_BATCH_SIZE	512	No	The maximum number of requests that are allowed in a batch in a thread to write data. If the total number of PUT requests reaches the upper limit that is specified by the WRITE_BATCH_SIZE parameter after conflicts are handled based on the WRITE_MODE parameter, the data is submitted in a batch.
     WRITE_BATCH_BYTE_SIZE	2 MB	No	The maximum number of bytes that are allowed in a batch in a thread to write data. If the total number of bytes of the PUT requests reaches the upper limit that is specified by the WRITE_BATCH_BYTE_SIZE parameter after conflicts are handled based on the WRITE_MODE parameter, the data is submitted in a batch.
     WRITE_MAX_INTERVAL_MS	10000 ms	No	The interval at which data is submitted in a batch.
     WRITE_FAIL_STRATEGY	TYR_ONE_BY_ONE	No	The policy that is used to handle submission failures. If a batch cannot be submitted, Holo Client submits the data entries one at a time in the batch in the specified sequence. If a data entry cannot be submitted, Holo Client returns an error message that contains the information about the data entry.
     WRITE_THREAD_SIZE	1	No	The number of parallel threads that are used to write data. Each thread occupies one connection. The total number of connections occupied by a Spark job varies based on the Spark parallelism. The total number of connections can be calculated by using the following formula: Total number of connections = Value of spark.default.parallelism × Value of WRITE_THREAD_SIZE.
     DYNAMIC_PARTITION	false	No	Specifies whether to automatically create a partition if the data is written to a parent table that has no partitions. Valid values: true and false. true: A partition is automatically created if the data is written to a parent table that has no partitions.
     RETRY_COUNT	3	No	The maximum number of retries allowed to write and query data if a connection failure occurs.
     RETRY_SLEEP_INIT_MS	1000 ms	No	The interval at which a retry is performed. The amount of time consumed by the retries for a request is calculated by using the following formula: Value of the RETRY_SLEEP_INIT_MS parameter + Value of the RETRY_COUNT parameter × Value of the RETRY_SLEEP_STEP_MS parameter.
     RETRY_SLEEP_STEP_MS	10*1000 ms	No	The amount of time required for a retry. The amount of time consumed by the retries for a request is calculated by using the following formula: Value of the RETRY_SLEEP_INIT_MS parameter + Value of the RETRY_COUNT parameter × Value of the RETRY_SLEEP_STEP_MS parameter.
     CONNECTION_MAX_IDLE_MS	60000 ms	No	The idle timeout period for the connections that are used to read and write data. If a connection remains idle for a period of time longer than the specified idle timeout period, Holo Client automatically releases the connection.
     FIXED_CONNECTION_MODE	false	No	Specifies whether to use fixed connections. In a non-fixed copy mode, such as the INSERT mode, data writes and point queries do not occupy connections. Note The fixed connection feature is in beta release and is available only when the connector version is V1.2.0 or later and the Hologres engine version is r1.3.0 or later.

Data type mappings

The following table describes data type mappings between Spark and Hologres.