Realtime Compute for Apache Flink:JARドラフトの開発

Apache Flinkの環境依存関係を構成する

次のサンプルコードは、Flink関連の依存関係の例を示しています。ログファイルに関連する依存関係を追加する必要がある場合もあります。完全な依存関係の詳細については、このトピックの「完全なサンプルコード」セクションを参照してください。

         <!-- Apache Flink dependencies -->
        <!-- Provide the dependencies to prevent them from being packaged in a JAR file.   -->
        <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コネクタを使用してRealtime Compute for Apache Flinkに接続する必要があります。Maven中央リポジトリに保存されている VVR DataStreamコネクタ を使用して、ドラフトを開発できます。

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

OSSに保存されている依存関係ファイルからデータを読み取る

Realtime Compute for Apache FlinkのJARデプロイメントのMain関数を使用して、ローカル構成ファイルからデータを読み取ることはできません。この場合、構成ファイルをObject Storage Service（OSS）バケットにアップロードし、JARデプロイメントを作成するときに [JARデプロイメントの作成] ダイアログボックスの [追加の依存関係] フィールドに構成ファイルを追加して、構成ファイルからデータを読み取ることができます。例：

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. デプロイメントの実行時に、追加の依存関係/flink/usrlib フィールドに追加された追加の依存関係ファイルは、デプロイメントが実行されているポッドの ディレクトリにロードされます。

   3. JARドラフトで次のコードを実行して、構成ファイルからデータを読み取ります。

                  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：Realtime Compute for Apache Flinkのワークスペースがアクセス許可を持つOSSバケットの構成ファイルからデータを読み取る

   1. 構成ファイルconfig.propertiesをOSSバケットにアップロードします。

   2. JARドラフトで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. 外部データソースをRealtime Compute for Apache Flinkのデータストリーミングプログラムに統合します。 Watermarks は、Realtime Compute for Apache Flinkの時間セマンティクスに基づく計算戦略です。ほとんどの場合、ウォーターマークはタイムスタンプと一緒に使用されます。この例では、ウォーターマーク戦略は使用されていません。詳細については、「Generating Watermarks」をご参照ください。

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

2. 演算子を変換します。この例では、DataStream<String> 演算子が DataStream<Student> に変換されます。より複雑な演算子を変換および処理する方法の詳細については、「演算子」をご参照ください。

             // データ構造が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); // scoreが60以上のデータレコードを取得します。

ドラフトのJARファイルをパッケージ化する

maven-shade-pluginを使用して、ドラフトのJARファイルをパッケージ化します。

<build>
        <plugins>
            <!-- Java compiler -->
            <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>