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 API でデータを読み書きするには、DataStream コネクタを使用して Realtime Compute for Apache Flink に接続します。Maven 中央リポジトリは、開発中に使用できる VVR DataStream コネクタを提供しています。

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

OSS からの追加依存関係の読み取り

Flink JAR ジョブは、main メソッドからローカル設定ファイルを読み取ることはできません。代わりに、設定ファイルをワークスペースの 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：ワークスペースバケットからの読み取り

   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：承認済みバケットからの読み取り

   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 の計算戦略であり、タイムスタンプと共に使用されることが多いため、この例ではウォーターマーク戦略を使用しません。詳細については、「ウォーターマーク戦略」をご参照ください。

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

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

             // データ構造を 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>