实时数仓Hologres：Hologres批量写入最佳实践

基础环境准备

您需准备以下环境：

• 开通V2.2.25或以上版本的Hologres实例，并创建数据库。

• 创建EMR-Spark集群（Spark版本需大于或等于3.3.0）。详情请参见创建集群。

• 下载Spark-Connector包。

  您可通过Maven中央仓库下载Spark读写Hologres时需要引用的连接器JAR包hologres-connector-spark-3.x。

本文中使用的环境信息如下：

测试数据准备

1. 准备原始数据。

   1. 登录EMR-Spark集群的master节点，详情请参见连接ECS实例的方式。

   2. 单击TPC-H_Tools_v3.0.0.zip下载TPCH工具，将其复制到集群master节点的ECS机器中进行解压，然后进入TPC-H_Tools_v3.0.0/TPC-H_Tools_v3.0.0/dbgen目录。

   3. 执行如下代码，在dbgen目录下生成1 TB测试集文件customer.tbl，原tbl格式文件大小为23 GB。

      ./dbgen -s 1000 -T c

      TPC-H 1 TB数据集中customer表的详细信息如下：

      表信息	说明
      字段数据量	8
      字段类型	INT、BIGINT、TEXT、DECIMAL
      数据行数	150,000,000
      表Shard数	40

2. 测试数据导入Spark。

   执行如下命令，将customer.tbl文件上传到Spark集群。

   hadoop fs -put customer.tbl <spark_resource>

   其中spark_resource为创建EMR-Spark集群时，配置的集群存储根路径。

3. 在Hologres中创建不同存储格式的表。SQL命令如下：

   行列共存

   CREATE TABLE test_table_mixed (
       C_CUSTKEY BIGINT PRIMARY KEY,
       C_NAME TEXT,
       C_ADDRESS TEXT,
       C_NATIONKEY INT,
       C_PHONE TEXT,
       C_ACCTBAL DECIMAL(15, 2),
       C_MKTSEGMENT TEXT,
       C_COMMENT TEXT
   )
   WITH (
       orientation = 'column,row'
   );

   列存（有主键）

   CREATE TABLE test_table_column (
       C_CUSTKEY BIGINT PRIMARY KEY,
       C_NAME TEXT,
       C_ADDRESS TEXT,
       C_NATIONKEY INT,
       C_PHONE TEXT,
       C_ACCTBAL DECIMAL(15, 2),
       C_MKTSEGMENT TEXT,
       C_COMMENT TEXT
   )
   WITH (
       orientation = 'column'
   );

   列存（无主键）

   CREATE TABLE test_table_column_no_pk (
       C_CUSTKEY BIGINT,
       C_NAME TEXT,
       C_ADDRESS TEXT,
       C_NATIONKEY INT,
       C_PHONE TEXT,
       C_ACCTBAL DECIMAL(15, 2),
       C_MKTSEGMENT TEXT,
       C_COMMENT TEXT
   )
   WITH (
       orientation = 'column'
   );

   行存（有主键）

   CREATE TABLE test_table_row (
       C_CUSTKEY BIGINT PRIMARY KEY,
       C_NAME TEXT,
       C_ADDRESS TEXT,
       C_NATIONKEY INT,
       C_PHONE TEXT,
       C_ACCTBAL DECIMAL(15, 2),
       C_MKTSEGMENT TEXT,
       C_COMMENT TEXT
   )
   WITH (
       orientation = 'row'
   );

   行存（无主键）

   CREATE TABLE test_table_row_no_pk (
       C_CUSTKEY BIGINT,
       C_NAME TEXT,
       C_ADDRESS TEXT,
       C_NATIONKEY INT,
       C_PHONE TEXT,
       C_ACCTBAL DECIMAL(15, 2),
       C_MKTSEGMENT TEXT,
       C_COMMENT TEXT
   )
   WITH (
       orientation = 'row'
   );

测试配置

本文主要测试不同模式下的导入性能。

1. 登录EMR-Spark集群的master节点（登录方式请参见连接ECS实例的方式），上传已下载的Spark-Connector包，然后执行如下命令进入spark-sql交互界面。

   说明

   通过调整spark.sql.files.maxPartitionBytes参数值， 可以控制Spark读取HDFS文件的并发。此处并发控制为40。

   # 进入spark-sql交互界面
   
   spark-sql --jars <path>/hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar \
   --conf spark.executor.instances=40 \
   --conf spark.executor.cores=1 \
   --conf spark.executor.memory=4g \
   --conf spark.sql.files.maxPartitionBytes=644245094

   其中path为hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar所在的根路径。

2. 在spark-sql交互界面执行如下SQL，通过创建临时表的方式写入数据。

   由于测试过程中，需要多次调整参数以测试不同模式的写入性能，此处选择创建临时表的方式。

   说明

   实际使用中，您可直接使用Catalog加载Hologres表，更加方便。

   -- 创建csv格式的临时表
   CREATE TEMPORARY VIEW csvtable (
     c_custkey BIGINT,
     c_name STRING,
     c_address STRING,
     c_nationkey INT,
     c_phone STRING,
     c_acctbal DECIMAL(15, 2),
     c_mktsegment STRING,
     c_comment STRING)
   USING csv OPTIONS (
     path "<spark_resources>/customer.tbl", sep "|"
   );
   
   CREATE TEMPORARY VIEW hologresTable (
     c_custkey BIGINT,
     c_name STRING,
     c_address STRING,
     c_nationkey INT,
     c_phone STRING,
     c_acctbal DECIMAL(15, 2),
     c_mktsegment STRING,
     c_comment STRING)
   USING hologres OPTIONS (
     jdbcurl "jdbc:postgresql://<hologres_vpc_endpoint>/<database_name>",
     username "<accesskey_id>", 
     password "<accesskey_secret>", 
     table "<table_name>",
     direct_connect "false",
     write.mode "auto",
     write.insert.thread_size "3",
     write.insert.batch_size "2048"
   );
   
   INSERT INTO hologresTable SELECT * FROM csvTable;

   参数说明如下：

   参数名	描述
   spark_resources	创建EMR-Spark集群时，配置的集群存储根路径。 您可登录EMR on ECS控制台，单击目标集群ID，在基础信息页面的集群信息区域获取集群存储根路径。
   hologres_vpc_endpoint	Hologres实例对应的VPC网络类型的域名。 您可登录Hologres管理控制台，单击目标实例ID，在实例详情页的网络信息区域，获取指定VPC的域名。例如：杭州地域的VPC Endpoint格式为<Instance ID>-cn-hangzhou-vpc-st.hologres.aliyuncs.com:80。
   database_name	Hologres实例的数据库名称。
   accesskey_id	具有Hologres相应Database读取权限的AccessKey ID。
   accesskey_secret	具有Hologres相应Database读取权限的AccessKey Secret。
   table_name	待写入的Hologres目标表名称。
   write.mode	写入模式，取值如下： auto ：默认值，Connector自行选择最佳的模式。 insert：使用INSERT INTO VALUES方式写入。 stream：使用FIXED COPY流式写入。 bulk_load ：使用COPY模式批量导入无主键表。 bulk_load_on_conflict ：使用COPY模式批量导入有主键表。
   write.insert.thread_size	写入并发，仅使用INSERT写入时生效。
   write.insert.batch_size	写入攒批，仅使用INSERT写入时生效。
   write.on_conflict_action	INSERT_OR_REPLACE（默认）：主键冲突更新。 INSERT_OR_IGNORE： 主键冲突忽略。

   更多参数信息请参见参数说明。

测试场景

测试结果

测试结果包括以下字段：

SELECT
    COUNT(*), AVG(duration) AS avg_duration_ms
FROM
    hologres.hg_query_log
WHERE
    query_start >= '<start_time>' AND query_start <= '<end_time>'
    AND query LIKE '%<test_table_column>%' AND command_tag = 'COPY';

详细测试结果如下：