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 ステートメントを使用して、外部テーブルのデータを読み書きします。

  -- テーブルから読み取ります。
  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 ジョブを開発し、write API を呼び出してデータを書き込むことができます。 ジョブは CSV ファイルからデータを読み取り、DataFrame に変換してから、Hologres インスタンスに書き込みます。 以降のセクションでは、さまざまなプログラミング言語のサンプルコードを提供します。 Hologres 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 でパッケージ化します。たとえば、出力 JAR ファイルが 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 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 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 API を呼び出してデータを DataFrame にロードできます。以下の例では、さまざまなプログラミング言語で Hologres テーブルから DataFrame にデータを読み取る方法を示します。Hologres 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().filter("c_custkey < 500")

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 に設定されている場合にのみ有効になります。

読み取りパラメータ