Realtime Compute for Apache Flink:Hive 方言ジョブ

前提条件

• RAM ユーザー、RAM ロール、またはその他のアイデンティティとしてコンソールにアクセスする場合、必要な権限が付与されていることを確認してください。詳細については、「権限」をご参照ください。

• ワークスペースが作成済みであること。詳細については、「Realtime Compute for Apache Flink の有効化」をご参照ください。

制限事項

• Hive 方言は、Ververica Runtime (VVR) 8.0.11 以降のバージョンでのみサポートされています。

• 現在、SQL ジョブは Hive 方言の INSERT Statements 構文のみをサポートしており、INSERT Statements の前に USE Catalog <yourHiveCatalog> を宣言する必要があります。テーブルを作成する必要がある場合は、[Scripts] ページで操作してください。

• Hive および Flink のユーザー定義関数 (UDF) はサポートされていません。

ステップ1：Hive カタログの作成

1. Hive メタデータを設定します。詳細については、「Hive メタデータの設定」をご参照ください。

2. Hive カタログを作成します。詳細については、「Hive カタログの作成」をご参照ください。

   このチュートリアルでは、Hive カタログの名前を hdfshive とします。

ステップ2：Hive サンプルテーブルの準備

1. 左側のナビゲーションペインで、[Development] > [Scripts] に移動し、[New] をクリックしてスクリプトを作成します。

2. 次のサンプル SQL ステートメントを実行します。

   重要

   Hive のソーステーブルとシンクテーブルは、CREATE TABLE ステートメントで作成された永続テーブルである必要があります。CREATE TEMPORARY TABLE ステートメントで作成された一時テーブルは使用できません。

   -- Hive カタログを使用します。この例では、カタログ名は hdfshive で、ステップ1で作成したものです。
   USE CATALOG hdfshive;   
   -- デフォルトのストレージ形式でソーステーブルを作成します。
   CREATE TABLE source_table (
    id INT,
    name STRING,
    age INT,
    city STRING,
    salary FLOAT
   )WITH ('connector' = 'hive');
   -- デフォルトのストレージ形式でシンクテーブルを作成します。
   CREATE TABLE target_table (
   city STRING,
   avg_salary FLOAT,
   user_count INT
   )WITH ('connector' = 'hive');
   -- ソーステーブルにサンプルデータを挿入します。
   INSERT INTO source_table VALUES
   (1, 'Alice', 25, 'New York', 5000.0),
   (2, 'Bob', 30, 'San Francisco', 6000.0),
   (3, 'Charlie', 35, 'New York', 7000.0),
   (4, 'David', 40, 'San Francisco', 8000.0),
   (5, 'Eva', 45, 'Los Angeles', 9000.0);
   -- 特定のストレージ形式 (例: Parquet) でテーブルを作成します。
   -- Hive モジュールをロードします。
   load MODULE hive with ('hive-version' = '2.3.6');
   use CATALOG `hdfshive`;
   -- 必須: 'STORED' などの Hive DDL キーワードを認識できるように、SQL 方言を 'hive' に設定します。
   set 'table.sql-dialect' = 'hive';
   CREATE TABLE `parquet_table`(
    id INT,
    name STRING,
    age INT,
    city STRING,
    salary FLOAT
   )STORED AS PARQUET;

ステップ3：Hive SQL ジョブの作成

1. 左側のナビゲーションペインで、[Development] > [ETL] に移動します。

2. [新規]をクリックします。[新規ドラフト]ダイアログボックスで、[空白のバッチドラフト (BETA)]を選択し、[次へ]をクリックします。

3. ジョブ情報を入力します。

   パラメーター	説明	例
   [名前]	ジョブの名前。 説明 ジョブ名は現在のワークスペース内で一意である必要があります。	hive-sql
   [場所]	ジョブのコードファイルが保存されるフォルダー。 既存のフォルダーの右側にある アイコンをクリックして、サブフォルダーを作成することもできます。	Drafts
   [エンジンバージョン]	ジョブで使用する Flink エンジンバージョン。 [推奨] タグが付いたバージョンを選択することをお勧めします。これらのバージョンは、信頼性とパフォーマンスがより高くなっています。エンジンバージョンの詳細については、「リリースノート」および「エンジンバージョン」をご参照ください。	vvr-8.0.11-flink-1.17
   [SQL 方言]	データ処理に使用する SQL 言語。 説明 このパラメーターは、Hive 方言をサポートするエンジンバージョンを選択した場合にのみ表示されます。	Hive SQL

4. [作成] をクリックします。

ステップ4：Hive SQL ジョブの作成とデプロイ

1. SQL ステートメントを作成します。

   この例では、30 歳以上のユーザー数と各都市の平均給与を計算します。次の SQL スクリプトを SQL エディターにコピーできます。

   -- Hive カタログを使用します。この例では、カタログ名は hdfshive で、ステップ1で作成したものです。
   USE CATALOG hdfshive; 
   INSERT INTO TABLE target_table
   SELECT
     city,
     AVG(salary) AS avg_salary, -- 平均給与を計算
     COUNT(id) AS user_count -- ユーザー数をカウント
   FROM source_table
   WHERE age > 30 -- 30 歳以上のユーザーをフィルタリング
   GROUP BY city; -- city でグループ化

2. 右上隅で、[デプロイ] をクリックします。 ダイアログボックスで、必要に応じてパラメーターを設定し (このチュートリアルではデフォルト設定を使用します)、[OK] をクリックします。

(オプション) ステップ5：ランタイムパラメーターの設定

1. 左側のナビゲーションペインで、[O&M] > [Deployments] に移動します。

2. ドロップダウンリストから、[バッチジョブ] を選択します。対象のジョブを見つけ、「操作」列の [詳細] をクリックします。

3. デプロイ詳細パネルで、ランタイム[ パラメータ設定] セクションの [編集] をクリックします。

4. [その他の設定] フィールドに、以下の設定を追加します。

   fs.oss.jindo.endpoint: <YOUR_Endpoint> 
   fs.oss.jindo.accessKeyId: <YOUR_AccessKeyId>
   fs.oss.jindo.accessKeySecret: <YOUR_AccessKeySecret>

   これらのパラメーターの詳細については、「OSS-HDFS へのデータ書き込み」をご参照ください。

5. [保存] をクリックします。

ステップ6：ジョブの開始と結果の表示

1. [デプロイ] ページで、フィルターから [バッチジョブ] を選択し、目的のジョブ (たとえば hive-sql) を見つけ、[アクション] 列の [開始] をクリックします。

2. ジョブのステータスが[完了]に変更されたら、結果を確認します。

   [開発] > [スクリプト] ページで、各都市の 30 歳を超えるユーザー数とその平均給与を含むデータを表示するには、次の SQL 文を実行します。

   -- Hive カタログを使用します。この例では、カタログ名は hdfshive で、ステップ1で作成したものです。
   USE CATALOG hdfshive; 
   select * from target_table;

   クエリは、target_table から city、avg_salary、user_count の列を含む 3 行を返します：Los Angeles (9000.0, 1)、New York (7000.0, 1)、San Francisco (8000.0, 1)。

Hive JAR ジョブの開発

Hive 方言ジョブは JAR ジョブとして実行できます。これには、「ververica-connector-hive-2.3.6」 JAR パッケージが必要です。また、JAR ジョブ内の Hive 設定とコンソールの設定が一致していることを確認する必要もあります。

1. コンソールの設定

   1. [JAR URI] には、JAR ジョブ用にアップロードされた JAR パッケージを指定します。

   2. [Additional Dependencies] で、Hive クラスターから取得した 4 つの設定ファイル (core-site.xml、mapred-site.xml、hdfs-site.xml、hive-site.xml) をアップロードします。また、ververica-connector-hive-2.3.6 の JAR パッケージもアップロードする必要があります。

   3. ランタイムパラメーターを設定します。Hive クラスターの設定に基づき、OSS-HDFS にデータを書き込む必要がある場合は、「(オプション) ステップ5：ランタイムパラメータの設定」で説明されている設定を使用します。

      table.sql-dialect: HIVE
      classloader.parent-first-patterns.additional: org.apache.hadoop;org.antlr.runtime
      kubernetes.application-mode.classpath.include-user-jar: true

2. JAR ジョブコードの例：

   1. StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
      Configuration conf = new Configuration();
      conf.setString("type", "hive");
      conf.setString("default-database", "default");
      conf.setString("hive-version", "2.3.6"); 
      conf.setString("hive-conf-dir", "/flink/usrlib/" );
      conf.setString("hadoop-conf-dir", "/flink/usrlib/");
      CatalogDescriptor descriptor = CatalogDescriptor.of("hivecat", conf);
      tableEnv.createCatalog("hivecat", descriptor);
      tableEnv.loadModule("hive", new HiveModule());
      tableEnv.useModules("hive");
      tableEnv.useCatalog("hivecat");
      tableEnv.executeSql("insert into `hivecat`.`default`.`test_write` select * from `hivecat`.`default`.`test_read`;");

関連ドキュメント

• Hive 方言の INSERT ステートメント構文の詳細については、「INSERT ステートメント｜Apache Flink」をご参照ください。

• Flink SQL を使用したバッチデータ処理の詳細については、「バッチ処理」をご参照ください。