DataStream コネクタを使用したデータソースの読み取りと書き込み - Realtime Compute for Apache Flink

このトピックでは、DataStream コネクタを使用してデータソースの読み取りと書き込みを行う方法を説明します。

DataStream コネクタの依存関係と使用方法

DataStream API を使用してデータソースの読み取りまたは書き込みを行うには、対応する DataStream コネクタを使用して Realtime Compute for Apache Flink に接続します。VVR DataStream コネクタは Maven 中央リポジトリで入手でき、ジョブ開発で直接使用できます。

重要

「サポートされているコネクタ」で、DataStream API をサポートすると明記されているコネクタのみを使用してください。このリストにないコネクタはサポートされておらず、そのインターフェイスは予告なく変更される場合があります。

DataStream コネクタは商用暗号化されているため、直接実行することはできません。ローカルでのデバッグと実行については、「コネクタを含むジョブをローカルで実行およびデバッグする」をご参照ください。

コネクタは、次のいずれかの方法で使用できます：

(推奨) 追加の依存関係

ジョブの Maven pom.xml ファイルで、必要なコネクタを provided スコープでプロジェクトの依存関係として追加します。
説明
- ${vvr.version} は、ジョブのランタイム環境のエンジンバージョンを指定します。たとえば、ジョブが vvr-8.0.9-flink-1.17 エンジンで実行される場合、その Flink バージョンは 1.17.2 です。最新のエンジンを使用することを推奨します。特定のバージョンに関する詳細については、「エンジンバージョン」をご参照ください。
- コネクタの JAR パッケージは追加の依存関係として追加されるため、この依存関係を JAR パッケージにバンドルする必要はありません。したがって、スコープを provided として宣言する必要があります。
```

  <dependency>
      <groupId>com.alibaba.ververica</groupId>
      <artifactId>ververica-connector-mysql</artifactId>
      <version>${vvr.version}</version>
      <scope>provided</scope>
  </dependency>
```

新しいコネクタを開発したり、既存のコネクタの機能を拡張したりする必要がある場合、プロジェクトには共通のコネクタパッケージ flink-connector-base または ververica-connector-common への依存関係も必要です。

<!-- Flink コネクタのパブリックインターフェイスの基本依存関係 -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-base</artifactId>
    <version>${flink.version}</version>
</dependency>
<!-- Alibaba Cloud コネクタのパブリックインターフェイスの基本依存関係 -->
<dependency>
    <groupId>com.alibaba.ververica</groupId>
    <artifactId>ververica-connector-common</artifactId>
    <version>${vvr.version}</version>
</dependency>

JAR ジョブをデプロイし、コネクタの JAR パッケージを [追加の依存関係] セクションに追加します。ご自身で開発したコネクタ、または Realtime Compute for Apache Flink が提供するコネクタをアップロードできます。

たとえば、ververica-connector-mysql-1.17-vvr-8.0.9.jar と ververica-connector-kafka-1.17-vvr-8.0.9.jar を追加します。

プロジェクトの依存関係

ジョブの Maven pom.xml ファイルで、必要なコネクタをプロジェクトの依存関係として追加します。次の例では、Kafka コネクタと MySQL コネクタを追加する方法を示します。
説明
- ${vvr.version} は、ジョブのランタイム環境のエンジンバージョンです。たとえば、ジョブが vvr-8.0.9-flink-1.17 エンジンで実行される場合、その Flink バージョンは 1.17.2 です。最新のエンジンを使用することを推奨します。詳細については、「エンジンバージョン」をご参照ください。
- コネクタはプロジェクトの依存関係としてジョブの JAR に直接パッケージ化されるため、デフォルトの compile スコープにする必要があります。
```
        
        <dependency>
            <groupId>com.alibaba.ververica</groupId>
            <artifactId>ververica-connector-kafka</artifactId>
            <version>${vvr.version}</version>
        </dependency>
        
        <dependency>
            <groupId>com.alibaba.ververica</groupId>
            <artifactId>ververica-connector-mysql</artifactId>
            <version>${vvr.version}</version>
        </dependency>
```

新しいコネクタを開発したり、既存のコネクタの機能を拡張したりする必要がある場合、プロジェクトには flink-connector-base または ververica-connector-common 共通コネクタパッケージへの依存関係も必要です。

        <!-- Flink コネクタのパブリックインターフェイス -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-base</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <!-- Alibaba Cloud コネクタのパブリックインターフェイスの基本依存関係 -->
        <dependency>
            <groupId>com.alibaba.ververica</groupId>
            <artifactId>ververica-connector-common</artifactId>
            <version>${vvr.version}</version>
        </dependency>

重要

依存関係の競合を防ぐには、次の点にご注意ください：

${flink.version} は、ジョブが実行される Flink のバージョンです。このバージョンは、ジョブのデプロイメントページで選択する VVR エンジンの Flink バージョンと同じである必要があります。たとえば、デプロイメントページで vvr-8.0.9-flink-1.17 エンジンを選択した場合、その Flink バージョンは 1.17.2 です。最新のエンジンを使用することを推奨します。特定のバージョンに関する詳細については、「エンジンバージョン」をご参照ください。
Flink 関連の依存関係については、依存関係に <scope>provided</scope> を追加して、スコープを provided に設定します。これには主に、org.apache.flink グループ内の flink- で始まるコネクタ以外の依存関係が含まれます。
Apache Flink のソースコードでは、@Public または @PublicEvolving のアノテーションが付いたメソッドのみを呼び出してください。Realtime Compute for Apache Flink は、これらのパブリック API の互換性のみを保証します。
DataStream API が組み込みの Flink コネクタでサポートされている場合は、その組み込みの依存関係を使用することを推奨します。

Realtime Compute for Apache Flink:DataStream コネクタの設定

DataStream コネクタの依存関係と使用方法

(推奨) 追加の依存関係

プロジェクトの依存関係

関連ドキュメント