翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

# Apache Flink での Apache Hudi の使用
<a name="tutorial-hudi-for-flink"></a>

Apache Hudi は、挿入、更新、アップサート、削除などのレコードレベルのオペレーションを備えたオープンソースのデータ管理フレームワークであり、データ管理とデータパイプライン開発を簡素化するために使用できます。Amazon S3 の効率的なデータ管理と組み合わせて Hudi を使用すると、リアルタイムでデータを取り込んで更新できます。Hudi は、データセットで実行したすべてのオペレーションのメタデータを保持するため、すべてのアクションはアトミックで一貫性があります。

Apache Hudi は、Amazon EMR リリース 7.2.0 以降の Apache Flink を搭載した Amazon EMR on EKS で使用できます。Apache Hudi ジョブを開始して送信する方法については、次の手順を参照してください。

## Apache Hudi ジョブを送信する
<a name="tutorial-hudi-for-flink-submit-jobs"></a>

Apache Hudi ジョブを送信する方法については、次の手順を参照してください。

1. という名前 AWS の Glue データベースを作成します`default`。

   ```
   aws glue create-database --database-input "{\"Name\":\"default\"}"
   ```

1. [Flink Kubernetes Operator SQL の例](https://github.com/apache/flink-kubernetes-operator/tree/main/examples/flink-sql-runner-example)に従って、`flink-sql-runner.jar` ファイルを構築します。

1. 次のような Hudi SQL スクリプトを作成します。

   ```
   CREATE CATALOG hudi_glue_catalog WITH (
   'type' = 'hudi',
   'mode' = 'hms',
   'table.external' = 'true',
   'default-database' = 'default',
   'hive.conf.dir' = '/glue/confs/hive/conf/',
   'catalog.path' = 's3://<hudi-example-bucket>/FLINK_HUDI/warehouse/'
   );
   
   USE CATALOG hudi_glue_catalog;
   CREATE DATABASE IF NOT EXISTS hudi_db;
   use hudi_db;
   
   CREATE TABLE IF NOT EXISTS hudi-flink-example-table(
       uuid VARCHAR(20),
       name VARCHAR(10),
       age INT,
       ts TIMESTAMP(3),
       `partition` VARCHAR(20)
   )
   PARTITIONED BY (`partition`)
   WITH (
     'connector' = 'hudi',
     'path' = 's3://<hudi-example-bucket>/hudi-flink-example-table',
     'hive_sync.enable' = 'true',
     'hive_sync.mode' = 'glue',
     'hive_sync.table' = 'hudi-flink-example-table',
     'hive_sync.db' = 'hudi_db',
     'compaction.delta_commits' = '1',
     'hive_sync.partition_fields' = 'partition',
     'hive_sync.partition_extractor_class' = 'org.apache.hudi.hive.MultiPartKeysValueExtractor',
     'table.type' = 'COPY_ON_WRITE'
   );
   
   EXECUTE STATEMENT SET
   BEGIN
   
   INSERT INTO hudi-flink-example-table VALUES
       ('id1','Alex',23,TIMESTAMP '1970-01-01 00:00:01','par1'),
       ('id2','Stephen',33,TIMESTAMP '1970-01-01 00:00:02','par1'),
       ('id3','Julian',53,TIMESTAMP '1970-01-01 00:00:03','par2'),
       ('id4','Fabian',31,TIMESTAMP '1970-01-01 00:00:04','par2'),
       ('id5','Sophia',18,TIMESTAMP '1970-01-01 00:00:05','par3'),
       ('id6','Emma',20,TIMESTAMP '1970-01-01 00:00:06','par3'),
       ('id7','Bob',44,TIMESTAMP '1970-01-01 00:00:07','par4'),
       ('id8','Han',56,TIMESTAMP '1970-01-01 00:00:08','par4');
   
   END;
   ```

1. Hudi SQL スクリプトと `flink-sql-runner.jar` ファイルを S3 の場所にアップロードします。

1. `FlinkDeployments` YAML ファイルで、`hudi.enabled` を `true` に設定します。

   ```
   spec:
     flinkConfiguration:
       hudi.enabled: "true"
   ```

1. YAML ファイルを作成して設定を実行します。この例では、ファイル名は `hudi-write.yaml` です。

   ```
   apiVersion: flink.apache.org/v1beta1
   kind: FlinkDeployment
   metadata:
     name: hudi-write-example
   spec:
     flinkVersion: v1_18
     flinkConfiguration:
       taskmanager.numberOfTaskSlots: "2"
       hudi.enabled: "true"
     executionRoleArn: "<JobExecutionRole>"
     emrReleaseLabel: "emr-7.12.0-flink-latest"
     jobManager:
       highAvailabilityEnabled: false
       replicas: 1
       resource:
         memory: "2048m"
         cpu: 1
     taskManager:
       resource:
         memory: "2048m"
         cpu: 1
     job:
       jarURI: local:///opt/flink/usrlib/flink-sql-runner.jar
       args: ["/opt/flink/scripts/hudi-write.sql"]
       parallelism: 1
       upgradeMode: stateless
     podTemplate:
       spec:
         initContainers:
           - name: flink-sql-script-download
             args: 
               - s3
               - cp
               - s3://<s3_location>/hudi-write.sql
               - /flink-scripts
             image: amazon/aws-cli:latest
             imagePullPolicy: Always
             resources: {}
             terminationMessagePath: /dev/termination-log
             terminationMessagePolicy: File
             volumeMounts:
               - mountPath: /flink-scripts
                 name: flink-scripts
           - name: flink-sql-runner-download
             args: 
               - s3
               - cp
               - s3://<s3_location>/flink-sql-runner.jar
               - /flink-artifacts
             image: amazon/aws-cli:latest
             imagePullPolicy: Always
             resources: {}
             terminationMessagePath: /dev/termination-log
             terminationMessagePolicy: File
             volumeMounts:
               - mountPath: /flink-artifacts
                 name: flink-artifact
         containers:
           - name: flink-main-container
             volumeMounts:
               - mountPath: /opt/flink/scripts
                 name: flink-scripts
               - mountPath: /opt/flink/usrlib
                 name: flink-artifact
         volumes:
           - emptyDir: {}
             name: flink-scripts
           - emptyDir: {}
             name: flink-artifact
   ```

1. Flink Hudi ジョブを [Flink Kubernetes オペレータ](https://docs.aws.amazon.com/emr/latest/EMR-on-EKS-DevelopmentGuide/jobruns-flink-kubernetes-operator.html)に送信します。

   ```
   kubectl apply -f hudi-write.yaml
   ```