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

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

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

DataStream API を使用してデータソースからの読み取りまたはデータソースへの書き込みを行うには、対応する DataStream コネクタを使用して Realtime Compute for Apache Flink (VVR) に接続します。VVR DataStream コネクタは Maven セントラルリポジトリで提供されており、ジョブ開発で直接使用できます。

重要

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

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

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

（推奨）コネクタの依存関係スコープを provided に設定する

ジョブの 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 パッケージを追加します。

方法 1：プラットフォームの組み込みコネクタを使用する（推奨）

説明

組み込みコネクタの使用を推奨します。追加の依存関係とバージョン管理が削減されます。プラットフォームへのパッチ適用やアップグレード時に、組み込みコネクタはプラットフォームと共に更新されるため、個別の更新は不要です。
VVR 11.2 以降でのみサポートされます。

JAR ジョブをデプロイし、[起動パラメーター] セクションの [その他の設定] に設定を追加します。

たとえば mysql や kafka など、複数の組み込みコネクタに依存させるには、以下の設定を使用します。各組み込みコネクタの名前については、コネクタのドキュメントをご参照ください。

pipeline.used-builtin-connectors: mysql;kafka

方法 2：コネクタの JAR パッケージを追加の依存関係として追加する

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

たとえば、ververica-connector-mysql-1.17-vvr-8.0.8.jar および ververica-connector-kafka-1.17-vvr-8.0.8.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- で始まる Connector 以外の依存関係が含まれます。
Apache Flink ソースコードでは、@Public または @PublicEvolving でアノテーションが付けられたメソッドのみを呼び出してください。Realtime Compute for Apache Flink は、これらの公開 API についてのみ互換性を保証します。
DataStream API が組み込み Flink コネクタでサポートされている場合は、対応する組み込み依存関係を使用することを推奨します。

Realtime Compute for Apache Flink:DataStream コネクタのセットアップ

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

（推奨）コネクタの依存関係スコープを provided に設定する

プロジェクトの依存関係

関連ドキュメント