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 の一時テーブルを宣言します。

   カタログを使用したデータの書き込み

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

   一時ビューを使用したデータの書き込み

   -- 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 for 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 モードを使用すると、データ書き込み用の一時テーブルが作成されます。書き込み操作が成功した後、一時テーブルが元のテーブルを置き換えます。Overwrite モードは必要な場合にのみ使用してください。

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

読み取りパラメーター