Hologres:Spark を使用して Hologres の読み書きを行う

Hologres カタログの初期化

Spark クラスターで spark-sql を起動し、Hologres コネクタをロードして、カタログパラメーターを指定します。

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 カタログの共通コマンド

• Hologres カタログのロード

  Spark の Hologres カタログは Hologres データベースに正確に対応しており、使用中に変更することはできません。

  USE hologres_external_test_db;

• すべての名前空間のクエリ

  Spark の名前空間は Hologres のスキーマに対応します。デフォルトのスキーマは `public` です。USE 命令を使用してデフォルトのスキーマを変更できます。

  -- Hologres カタログ内のすべての名前空間を表示します。これは Hologres 内のすべてのスキーマに対応します。
  SHOW NAMESPACES;

• 名前空間内のテーブルのクエリ

  • すべてのテーブルのクエリ

    SHOW TABLES;

  • 特定の名前空間内のテーブルのクエリ

    USE test_schema;
    SHOW TABLES;
    
    -- または
    SHOW TABLES IN test_schema;

• テーブルの読み書き

  SELECT 文と INSERT 文を使用して、カタログ内の Hologres 外部テーブルの読み書きを行います。

  -- テーブルから読み取ります。
  SELECT * FROM public.test_table1;
  
  -- テーブルに書き込みます。
  INSERT INTO test_schema.test_table3 SELECT * FROM public.test_table1;

Spark-SQL を使用したデータのインポート

Spark-SQL を使用する場合、カタログを使用して Hologres テーブルのメタデータをロードする方が便利です。一時テーブルを作成して Hologres テーブルを宣言することもできます。

1. Hologres カタログを初期化します。

   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 カタログをロードします。
   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 ソースのスキーマ。
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 ソースのスキーマ。
        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 = 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 を使用する場合、カタログを使用して Hologres テーブルのメタデータをロードする方が便利です。一時テーブルを作成して Hologres テーブルを宣言することもできます。

1. Hologres カタログを初期化します。

   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 からデータを読み取ります。

   • カタログを使用してデータを読み取ります。

     -- Hologres カタログをロードします。
     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 テーブルの読み取りに使用するタスクの最大数。
       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 テーブルの読み取りに使用するタスクの最大数。
    .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 テーブルの読み取りに使用するタスクの最大数。
                .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 コネクタは Spark の SaveMode パラメーターをサポートしています。SQL の場合、これは INSERT INTO または INSERT OVERWRITE を意味します。DataFrame の場合、これは書き込み時に SaveMode を Append または Overwrite に設定することを意味します。Overwrite は書き込み用の一時テーブルを作成し、書き込みが成功した後に元のテーブルを置き換えます。注意して使用してください。

次のパラメーターは、write.mode が insert に設定されている場合にのみ有効です。

読み取りパラメーター