Hologres：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寫入

   -- 載入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;

   TEMPORARY VIEW寫入

   -- 建立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 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(query)

     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時生效。

讀取參數