Spark读写Hologres

Hologres Catalog初始化

在Spark集群中启动spark-sql，加载Hologres Connector并指定Catalog参数。

spark-sql --jars hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar \
--conf spark.sql.catalog.hologres_external_test_db=com.alibaba.hologres.spark3.HoloTableCatalog \
--conf spark.sql.catalog.hologres_external_test_db.username=*** \
--conf spark.sql.catalog.hologres_external_test_db.password=*** \
--conf spark.sql.catalog.hologres_external_test_db.jdbcurl=jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/test_db

Hologres Catalog常用命令

• 加载Hologres Catalog

  Spark中的Hologres Catalog完全对应一个Hologres的Database，使用过程中无法更改。

  USE hologres_external_test_db;

• 查询所有Namespace。

  Spark中的Namespace，对应Hologres中的Schema，默认为public，使用过程中可以使用USE指令调整默认的Schema。

  -- 查看Hologres Catalog中的所有Namespace, 即Hologres中所有的Schema。
  SHOW NAMESPACES;

• 查询Namespace下的表。

  • 查询所有表。

    SHOW TABLES;

  • 查询指定Namespace下的表。

    USE test_schema;
    SHOW TABLES;
    
    -- 或者使用 
    SHOW TABLES IN test_schema;

• 读取和写入表。

  使用SELECT和INSERT语句对Catalog中的Hologres外部表进行读写。

  -- 读取表。
  SELECT * FROM public.test_table1;
  
  -- 写入表。
  INSERT INTO test_schema.test_table3 SELECT * FROM public.test_table1;

使用Spark-SQL导入

在使用Spark-SQL时，借助Catalog加载Hologres表的元数据更加便捷，同时也可以通过创建临时表的方式来声明一个Hologres表。

1. 初始化Hologres Catalog。

   spark-sql --jars hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar \
   --conf spark.sql.catalog.hologres_external_test_db=com.alibaba.hologres.spark3.HoloTableCatalog \
   --conf spark.sql.catalog.hologres_external_test_db.username=*** \
   --conf spark.sql.catalog.hologres_external_test_db.password=*** \
   --conf spark.sql.catalog.hologres_external_test_db.jdbcurl=jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/test_db

2. 从CSV源表导入数据至Hologres外表。

   说明

   Spark的INSERT INTO语法不支持通过column_list指定部分列进行写入，例如使用INSERT INTO hologresTable(c_custkey) SELECT c_custkey FROM csvTable来表示只写入c_custkey这一个字段。

   若您希望写入部分所需字段，可以使用CREATE TEMPORARY VIEW 的方式声明仅所需字段的Hologres临时表。

   CATALOG写入

   TEMPORARY VIEW写入

   -- 加载Hologres Catalog。
   USE hologres_external_test_db;
   
   -- 创建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 "resources/customer", sep "," -- 本地测试，直接用文件所在绝对路径。
   );
   
   -- 将csv表的数据写入到Hologres中
   INSERT INTO public.customer_holo_table SELECT * FROM csvTable;

   -- 创建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 "resources/customer", sep ","
   );
   
   -- 创建hologres临时表
   CREATE TEMPORARY VIEW hologresTable (
       c_custkey BIGINT,
       c_name STRING,
       c_phone STRING)
   USING hologres OPTIONS (
       jdbcurl "jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/test_db",
       username "***", 
       password "***", 
       table "customer_holo_table"
   );
   
   INSERT INTO hologresTable SELECT c_custkey,c_name,c_phone FROM csvTable;

使用DataFrame导入

使用spark-shell、pyspark等开发Spark作业，您也可以调用DataFrame的write接口来进行写入。不同的开发语言会将读取到CSV文件数据转为DataFrame后，再写入Hologres实例，相关示例代码如下。Hologres-Connector-Spark更多参数信息，详情请参见参数说明。

import org.apache.spark.sql.types._
import org.apache.spark.sql.SaveMode

// csv源的schema
val schema = StructType(Array(
  StructField("c_custkey", LongType),
  StructField("c_name", StringType),
  StructField("c_address", StringType),
  StructField("c_nationkey", IntegerType),
  StructField("c_phone", StringType),
  StructField("c_acctbal", DecimalType(15, 2)),
  StructField("c_mktsegment", StringType),
  StructField("c_comment", StringType)
))

// 从csv文件读取数据为DataFrame
val csvDf = spark.read.format("csv").schema(schema).option("sep", ",").load("resources/customer")

// 将读取到的DataFrame写入到Hologres中
csvDf.write
.format("hologres")
.option("username", "***")
.option("password", "***")
.option("jdbcurl", "jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/test_db")
.option("table", "customer_holo_table")
.mode(SaveMode.Append)
.save()

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.*;
import org.apache.spark.sql.SaveMode;
import java.util.Arrays;
import java.util.List;

public class SparkTest {
    public static void main(String[] args) {
        // csv源的schema
        List<StructField> asList =
                Arrays.asList(
                        DataTypes.createStructField("c_custkey", DataTypes.LongType, true),
                        DataTypes.createStructField("c_name", DataTypes.StringType, true),
                        DataTypes.createStructField("c_address", DataTypes.StringType, true),
                        DataTypes.createStructField("c_nationkey", DataTypes.IntegerType, true),
                        DataTypes.createStructField("c_phone", DataTypes.StringType, true),
                        DataTypes.createStructField("c_acctbal", new DecimalType(15, 2), true),
                        DataTypes.createStructField("c_mktsegment", DataTypes.StringType, true),
                        DataTypes.createStructField("c_comment", DataTypes.StringType, true));
        StructType schema = DataTypes.createStructType(asList);

        // 使用本地模式运行
        SparkSession spark = SparkSession.builder()
                .appName("Spark CSV Example")
                .master("local[*]") 
                .getOrCreate();

        // 从csv文件读取数据为DataFrame
        // 本地测试采用customer数据的绝对路径
        Dataset<Row> csvDf = spark.read().format("csv").schema(schema).option("sep", ",").load("resources/customer");

        // 将读取到的DataFrame写入到Hologres中
        csvDf.write.format("hologres").option(
           "username", "***").option(
           "password", "***").option(
           "jdbcurl", "jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/test_db").option(
           "table", "customer_holo_table").mode(
           "append").save()
    }
}

<dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-sql_2.13</artifactId>
      <version>3.5.4</version>
      <scope>provided</scope>
</dependency>

from pyspark.sql.types import *

# csv源的schema
schema = StructType([
    StructField("c_custkey", LongType()),
    StructField("c_name", StringType()),
    StructField("c_address", StringType()),
    StructField("c_nationkey", IntegerType()),
    StructField("c_phone", StringType()),
    StructField("c_acctbal", DecimalType(15, 2)),
    StructField("c_mktsegment", StringType()),
    StructField("c_comment", StringType())
])

# 从csv文件读取数据为DataFrame
csvDf = spark.read.csv("resources/customer", header=False, schema=schema, sep=',')

# 将读取到的DataFrame写入到Hologres中
csvDf.write.format("hologres").option(
    "username", "***").option(
    "password", "***").option(
    "jdbcurl", "jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/test_db").option(
    "table", "customer_holo_table").mode(
    "append").save()

Spark执行不同语言的作业，具体操作如下：

• Scala

  • 您可以用示例代码生成sparktest.scala文件，通过如下方式执行作业。

    -- 加载依赖 
    spark-shell --jars hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar
    
    -- 本地测试使用绝对路径加载文件
    scala> :load D:/sparktest.scala

  • 您也可以在加载依赖完成后，直接将示例代码粘贴进去执行。

• Java

  您可以使用开发工具引入示例代码，通过Maven工具完成打包。例如包名spark_test.jar。通过下方代码执行作业。

  -- 作业jar包采用绝对路径
  spark-submit --class SparkTest  --jars hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar  D:\spark_test.jar

• Python

  您可以在下方代码执行完成后，直接将示例代码粘贴进去执行。

  pyspark --jars hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar

使用Spark-SQL读取

使用Spark-SQL时，通过Catalog加载Hologres表的元数据更加方便，您也可以通过创建临时表的方式声明一个Hologres表。

1. 初始化Hologres Catalog。

   spark-sql --jars hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar \
   --conf spark.sql.catalog.hologres_external_test_db=com.alibaba.hologres.spark3.HoloTableCatalog \
   --conf spark.sql.catalog.hologres_external_test_db.username=*** \
   --conf spark.sql.catalog.hologres_external_test_db.password=*** \
   --conf spark.sql.catalog.hologres_external_test_db.jdbcurl=jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/test_db

2. 读取Hologres数据。

   • 通过Catalog读取。

     -- 加载Hologres Catalog。
     USE hologres_external_test_db;
     
     -- 读取Hologres表，支持字段裁剪和谓词下推。
     SELECT c_custkey,c_name,c_phone FROM public.customer_holo_table WHERE c_custkey < 500 LIMIT 10;

   • 通过创建临时表读取。

     CREATE TEMPORARY VIEW(table)

     CREATE TEMPORARY VIEW(query)

     CREATE TEMPORARY VIEW hologresTable
     USING hologres OPTIONS (
       jdbcurl "jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/test_db",
       username "***", 
       password "***", 
       read.max_task_count "80", // Hologres表最多分为多少个task进行读取
       table "customer_holo_table"
     );
     
     -- 支持字段裁剪和谓词下推
     SELECT c_custkey,c_name,c_phone FROM hologresTable WHERE c_custkey < 500 LIMIT 10;

     CREATE TEMPORARY VIEW hologresTable
     USING hologres OPTIONS (
       jdbcurl "jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/test_db",
       username "***", 
       password "***", 
       read.query "select c_custkey,c_name,c_phone from customer_holo_table where c_custkey < 500 limit 10"
     );
     
     SELECT * FROM hologresTable LIMIT 5;

读取Hologres数据为DataFrame

使用spark-shell、pyspark等开发Spark作业，可以调用Spark的Read接口将数据读取为DataFrame以进行后续的计算。不同语言读取Hologres表为DataFrame的示例如下。Hologres-Connector-Spark更多参数信息，详情请参见参数说明。

val readDf = (
  spark.read
    .format("hologres")
    .option("username", "***")
    .option("password", "***")
    .option("jdbcurl", "jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/test_db")
    .option("table", "customer_holo_table")
    .option("read.max_task_count", "80") // Hologres表最多分为多少个task进行读取
    .load()
    .filter("c_custkey < 500")
)

readDf.select("c_custkey", "c_name", "c_phone").show(10)

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

public class SparkSelect {
    public static void main(String[] args) {
        
        // 使用本地模式运行
        SparkSession spark = SparkSession.builder()
                .appName("Spark CSV Example")
                .master("local[*]") 
                .getOrCreate();
                
        Dataset<Row> readDf = (
           spark.read
                .format("hologres")
                .option("username", "***")
                .option("password", "***")
                .option("jdbcurl", "jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/test_db")
                .option("table", "customer_holo_table")
                .option("read.max_task_count", "80") // Hologres表最多分为多少个task进行读取
                .load()
                .filter("c_custkey < 500")
        );
        readDf.select("c_custkey", "c_name", "c_phone").show(10);
    }
}

<dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-sql_2.13</artifactId>
      <version>3.5.4</version>
      <scope>provided</scope>
</dependency>

readDf = spark.read.format("hologres").option(
"username", "***").option(
"password", "***").option(
"jdbcurl", "jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/test_db").option(
"table", "customer_holo_table").option(
"read.max_task_count", "80").load()

readDf.select("c_custkey", "c_name", "c_phone").show(10)

通用参数

写入参数

Hologres Connector支持Spark的SaveMode参数。对SQL来说，即INSERT INTO或者 INSERT OVERWRITE。对DataFrame来说，即WRITE时设置SaveMode为Append或者Overwrite。其中Overwrite会创建临时表进行写入，并在写入成功之后替换原始表，请在必要时使用。

以下参数仅write.mode为insert时生效。

读取参数