Realtime Compute for Apache Flink:JAR ジョブの開発

Flink 環境の依存関係の設定

以下のコードは、いくつかの基本的な Flink の依存関係を示しています。ログファイルの依存関係を追加する必要がある場合もあります。依存関係の完全なリストについては、このトピックの最後にある「完全なサンプルコード」をご参照ください。

         <!-- Apache Flink の依存関係 -->
        <!-- これらの依存関係は JAR ファイルにパッケージ化すべきではないため、provided に設定します。 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>

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

DataStream を使用してデータを読み書きするには、対応する DataStream コネクタを使用する必要があります。Maven 中央リポジトリには、ジョブ開発中に直接使用できる VVR DataStream コネクタ が含まれています。

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

OSS からの追加の依存関係ファイルの読み取り

Flink JAR ジョブは、`main` 関数内からローカル設定ファイルを読み取ることをサポートしていません。代わりに、設定ファイルをご利用の Flink ワークスペースに関連付けられた OSS バケットにアップロードできます。ジョブをデプロイする際に、追加の依存関係として追加することでファイルを読み取ることができます。以下のセクションで例を示します。

1. コード内でプレーンテキストのパスワードを使用しないように、`config.properties` という名前の設定ファイルを作成します。

   # Kafka 
   bootstrapServers=host1:9092,host2:9092,host3:9092
   inputTopic=topic
   groupId=groupId
   # MySQL
   database.url=jdbc:mysql://localhost:3306/my_database
   database.username=username
   database.password=password

2. JAR ジョブ内のコードを使用して、OSS バケットに保存されている `config.properties` ファイルを読み取ります。

   方法1：ワークスペースにアタッチされた OSS バケットからの読み取り

   1. Realtime Compute for Apache Flink 開発コンソールで、左側のナビゲーションウィンドウの [リソース管理] ページに移動し、ファイルをアップロードします。

   2. ジョブの実行時に、デプロイメント中に追加した追加の依存関係ファイルは、ジョブが実行される Pod の /flink/usrlib ディレクトリにロードされます。

   3. 以下のコードは、設定ファイルを読み取る方法の例です。

                  Properties properties = new Properties();
                  Map<String,String> configMap = new HashMap<>();
      
                  try (InputStream input = new FileInputStream("/flink/usrlib/config.properties")) {
                      // プロパティファイルをロードします。
                      properties.load(input);
                      // プロパティ値を取得します。
                      configMap.put("bootstrapServers",properties.getProperty("bootstrapServers")) ;
                      configMap.put("inputTopic",properties.getProperty("inputTopic"));
                      configMap.put("groupId",properties.getProperty("groupId"));
                      configMap.put("url",properties.getProperty("database.url")) ;
                      configMap.put("username",properties.getProperty("database.username"));
                      configMap.put("password",properties.getProperty("database.password"));
                  } catch (IOException ex) {
                      ex.printStackTrace();
                  }

   方法2：ワークスペースがアクセス権を持つ OSS バケットからの読み取り

   1. 設定ファイルを宛先の OSS バケットにアップロードします。

   2. `OSSClient` を使用して、OSS に保存されているファイルを直接読み取ることができます。詳細については、「ストリーム」および「アクセス認証情報の管理」をご参照ください。以下はサンプルコードです。

      OSS ossClient = new OSSClientBuilder().build("Endpoint", "AccessKeyId", "AccessKeySecret");
      try (OSSObject ossObject = ossClient.getObject("examplebucket", "exampledir/config.properties");
           BufferedReader reader = new BufferedReader(new InputStreamReader(ossObject.getObjectContent()))) {
          // ファイルを読み込んで処理...
      } finally {
          if (ossClient != null) {
              ossClient.shutdown();
          }
      }

ビジネスコードの記述

1. 以下のコードは、外部データソースを Flink データストリームプログラムに統合する方法を示しています。ウォーターマークは、イベント時間の進行を測定する Flink のメカニズムであり、タイムスタンプと共によく使用されます。この例では、ウォーターマークポリシーは使用しません。詳細については、「Watermark Strategies」をご参照ください。

            // 外部データソースを Flink データストリームプログラムに統合します。
           // WatermarkStrategy.noWatermarks() は、ウォーターマークポリシーを使用しないことを示します。
           DataStreamSource<String> stream = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka Source");

2. 以下のコードは、オペレーター変換を示しています。この例では、DataStream<String> を DataStream<Student> に変換します。複雑なオペレーター変換と処理方法の詳細については、「Flink Operators」をご参照ください。

             // データ構造を student に変換するオペレーター。
             DataStream<student> source = stream
                   .map(new MapFunction<String, student>() {
                       @Override
                       public student map(String s) throws Exception {
                           // データはカンマで区切られます。
                           String[] data = s.split(",");
                           return new student(Integer.parseInt(data[0]), data[1], Integer.parseInt(data[2]));
                       }
                   }).filter(student -> student.score >=60); // データをフィルタリングして、スコアが 60 以上のものを見つけます。

ジョブのパッケージ化

`maven-shade-plugin` プラグインを使用してジョブをパッケージ化できます。

<build>
        <plugins>
            <!-- Java コンパイラ -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.11.0</version>
                <configuration>
                    <source>${target.java.version}</source>
                    <target>${target.java.version}</target>
                </configuration>
            </plugin>

            <!-- maven-shade-plugin を使用して、必要なすべての依存関係を含む fat jar を作成します。 -->
            <!-- プログラムのエントリポイントが変更された場合は、<mainClass> の値を変更してください。 -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.2.0</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <!-- 不要な依存関係をいくつか削除します。 -->
                        <configuration>
                            <artifactSet>
                                <excludes>
                                    <exclude>org.apache.flink:force-shading</exclude>
                                    <exclude>com.google.code.findbugs:jsr305</exclude>
                                    <exclude>org.slf4j:*</exclude>
                                    <exclude>org.apache.logging.log4j:*</exclude>
                                </excludes>
                            </artifactSet>
                            <filters>
                                <filter>
                                    <!-- META-INF フォルダ内の署名をコピーしないでください。そうしないと、JAR ファイルを使用する際にセキュリティ例外が発生する可能性があります。 -->
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                            <transformers>
                                <transformer
                                        implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass>com.aliyun.FlinkDemo</mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>