Hologres：Spark即時同步

前提條件

• 執行個體版本需為V0.9及以上版本。請在Hologres管控台的執行個體詳情頁查看當前執行個體版本，如執行個體是V0.9以下版本，請您使用自助升級或加入HologresDingTalk交流群反饋，詳情請參見如何擷取更多的線上支援？。

• 需要安裝對應版本的Spark環境，能夠運行spark-shell命令。

串連數使用

Hologres Spark Connector在進行讀寫時，會使用一定的JDBC串連數。可能受到如下因素影響：

• Spark的並發特性，在作業運行過程中，可以通過Spark UI觀察到並存執行的Task數量。

• Connector在操作時，對於固定複製（fixed copy）方式的寫入，每個並行作業都只使用一個JDBC串連。而採用INSERT方式寫入時，每個並發則會利用write_thread_size數量的JDBC串連。在進行讀取操作時，每個並發同樣使用一個JDBC串連。

• 其他方面可能使用的串連數：在作業啟動時，可能會有擷取Schema資訊的操作，這可能會短暫地建立1個串連。

因此作業使用的總串連數可以通過如下公式計算：

• fixed copy模式：parallelism*1+1

• 普通INSERT模式：parallelism*write_thread_size+1

通過Spark Connector寫入（推薦使用）

Hologres當前支援使用內建的Spark Connector將Spark資料寫入Hologres，相比其他寫入方式，調用基於Holo Client實現Connector寫入的方式效能更優。具體操作步驟如下，阿里雲也為您提供了相關的使用樣本，詳情請參見通過Spark Connector寫入使用樣本。

通過Spark讀取資料來源資料並寫入Hologres

1. Spark從資料來源讀取資料。

   Spark支援從不同資料來源讀取資料，具體資料來源分類如下。

   • Spark支援以Hologres為資料來源。

     Hologres相容PostgreSQL，因為Spark可以用讀取PostgreSQL的方式讀取Hologres中的資料。讀取代碼如下。

     說明

     在使用JDBC方式進行讀取前，請前往官網下載Postgresql JDBC Jar，本文以postgresql-42.2.18版本為例，在spark-shell啟動時執行./bin/spark-shell --jars /path/to/postgresql-42.2.18.jar載入該jar，可以與hologres-connector的jar包一同載入。

     // Read from some table, for example: tb008
     val readDf = spark.read
       .format("jdbc") //使用postgresql jdbc driver讀取holo
       .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()

     Spark Connector從V1.3.2版本開始支援讀取Hologres，並提供了最佳化的並發讀取功能。相較於使用PostgreSQL JDBC驅動的方法，還可以設定讀取的並發參數，根據Hologres表的Shard進行分區，從而實現並發讀取，顯著提升了效能。樣本如下。

     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) // 可選，如果不指定schema，預設讀取holo表全部欄位
     .option("username", "your_username")
     .option("password", "your_password")
     .option("jdbcurl", "jdbc:postgresql://hologres_endpoint/test_db")
     .option("table", "tb008")
     .option("scan_parallelism", "10") //讀取Hologres時的預設並發數，最大為holo表的shardcount
     .load()

   • Spark支援其他資料來源（如Parquet格式的檔案）。

     Spark支援從其他資料來源中讀取資料寫入Hologres中，例如使用Spark從Hive中讀取資料，代碼如下。

     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. Spark將讀到的資料寫入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) -- 寫入攢批大小
     .option(SourceProvider.INPUT_DATA_SCHEMA_DDL, df.schema.toDDL) -- 僅spark3.x需要
     .mode(SaveMode.Append) // 僅spark3.x需要
     .save()

通過Spark即時寫入資料至Hologres

1. 在Hologres建立一張表，用於接收資料，建立代碼如下。

   CREATE TABLE test_table_stream
   (
       value text,
       count bigint
   );

2. 讀取本地連接埠輸入行，進行詞頻統計並即時寫入Hologres中，相關範例程式碼如下。

   • 代碼：

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

   • 參數釋義：

     參數名	預設值	是否必填	參數描述
     username	無	是	登入Hologres帳號的AccessKey ID。您可以單擊AccessKey 管理來擷取。 建議您使用環境變數的方式調用使用者名稱和密碼，降低密碼泄露風險。
     password	無	是	登入Hologres帳號的AccessKey Secret。您可以單擊AccessKey 管理來擷取。 建議您使用環境變數的方式調用使用者名稱和密碼，降低密碼泄露風險。
     table	無	是	Hologres用於接收資料的表名稱。
     endpoint	無	與JDBCURL二選一	Hologres執行個體的網路網域名稱。 您可以進入Hologres管理主控台執行個體詳情頁，從網路資訊擷取主機和連接埠號碼。
     database	無	與JDBCURL二選一	Hologres接收資料的表所在資料庫名稱。
     jdbcurl	無	與ENDPOINT+DATABASE組合設定二選一	Hologres的JDBCURL。
     copy_write_mode	true	否	是否使用Fixed Copy方式寫入，Fixed Copy是Hologres V1.3新增的能力，相比INSERT方法，Fixed Copy方式可以更高的吞吐（因為是流模式），更低的資料延時，更低的用戶端記憶體消耗（因為不攢批）。 說明 需要Connector為V1.3.0及以上版本，Hologres引擎版本為V1.3.34及以上版本。
     copy_write_format	false	否	僅Copy模式生效，是否進行髒資料校正，開啟之後如果有髒資料，可以定位到寫入失敗的具體行。 說明 RecordChecker會對寫入效能造成一定影響，非排查環節不建議開啟。
     bulk_load	true	否	是否採用批量Copy方式寫入（與fixed copy不同，fixed copy是流式的）。 說明 Hologre V2.1版本對無主鍵表的寫入效能進行了最佳化。在Hologre V2.1版本中，無主鍵表的批量寫入操作不再會導致表鎖，而是採用了行鎖機制。這可以使其能夠與Fixed Plan並存執行，從而提高了資料處理的效率和並發性。 Connector為V1.4.0及以上版本，Hologres引擎需要V2.1.0及以上版本。
     max_cell_buffer_size	20971520（20 MB）	否	使用Copy模式寫入時，單個欄位的最大長度。
     copy_write_dirty_data_check	false	否	是否進行髒資料校正，開啟之後如果有髒資料，可以定位到寫入失敗的具體行，RecordChecker會對寫入效能造成一定影響，非排查環節不建議開啟。 說明 僅Copy模式生效。
     copy_write_direct_connect	對於可以直連的環境會預設使用直連。	否	僅Copy模式生效，Copy的瓶頸往往是VPC Endpoint的網路吞吐，因此Hologres會測試當前環境能否直連holo fe，如果支援則預設使用直連。此參數設定為false表示不使用直連。
     input_data_schema_ddl	無	spark3.x必填，值為<your_DataFrame>.schema.toDDL。	Spark中DataFrame的DDL。
     write_mode	INSERT_OR_REPLACE	否	當INSERT目標表為有主鍵的表時採用不同策略。 INSERT_OR_IGNORE：當主鍵衝突時，不寫入。 INSERT_OR_UPDATE：當主鍵衝突時，更新相應列。 INSERT_OR_REPLACE：當主鍵衝突時，更新所有列。
     write_batch_size	512	否	每個寫入線程的最大批次大小，在經過write_mode合并後的Put數量達到write_batch_size時進行一次批量提交。
     write_batch_byte_size	2 MB	否	每個寫入線程的最大批次Byte大小，在經過WRITE_MODE合并後的Put資料位元組數達到WRITE_BATCH_BYTE_SIZE時進行一次批量提交。
     write_max_interval_ms	10000 ms	否	距離上次提交超過write_max_interval_ms會觸發一次批量提交。
     write_fail_strategy	TYR_ONE_BY_ONE	否	當某一批次提交失敗時，會將批次內的記錄逐條提交（保序），單條提交失敗的記錄將會跟隨異常被拋出。
     write_thread_size	1	否	寫入並發線程數（每個並發佔用1個資料庫連接）。 在一個Spark作業中，佔用的總串連數與Spark並發相關，關係為總串連數= spark.default.parallelism * WRITE_THREAD_SIZE。
     dynamic_partition	false	否	若為true，寫入分區表父表時，當分區不存在時自動建立分區。
     retry_count	3	否	當串連故障時，寫入和查詢的重試次數。
     retry_sleep_init_ms	1000 ms	否	每次重試的等待時間=retry_sleep_init_ms+retry_count*retry_sleep_step_ms。
     retry_sleep_step_ms	10*1000 ms	否	每次重試的等待時間=retry_sleep_init_ms+retry_count*retry_sleep_step_ms。
     connection_max_idle_ms	60000 ms	否	寫入線程和點查線程資料庫連接的最大IDLE時間，超過此時間的串連將被釋放。
     fixed_connection_mode	false	否	非Copy模式（如INSERT預設）下，寫入和點查情境不佔用串連數。 說明 Beta功能，需要Connector版本為V1.2.0及以上版本，Hologres引擎版本為V1.3.0及以上版本。
     scan_batch_size	256	否	在從Hologres讀取資料時，Scan操作一次擷取的行數。
     scan_timeout_seconds	60	否	在從Hologres讀取資料時，Scan操作的逾時時間，單位：秒（s）。
     scan_parallelism	10	否	讀取Hologres時的分區數量，最大為Hologres表的Shardcount。作業運行時，這些分區會被分配到Spark Task上進行讀取。

資料類型映射

Spark與Hologres的資料類型映射如下表所示。