Realtime Compute for Apache Flink：Mengembangkan pekerjaan JAR

Konfigurasikan dependensi lingkungan Flink

Kode berikut menunjukkan beberapa dependensi dasar Flink. Anda mungkin juga perlu menambahkan dependensi untuk file log. Untuk daftar lengkap dependensi, lihat Kode contoh lengkap di akhir topik ini.

         <!-- Apache Flink Dependencies -->
        <!-- These dependencies are provided because they should not be packaged into the 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>

Dependensi dan penggunaan konektor

Untuk membaca dan menulis data menggunakan DataStream, Anda harus menggunakan konektor DataStream yang sesuai. Repositori pusat Maven berisi konektor DataStream VVR yang dapat Anda gunakan langsung selama pengembangan pekerjaan.

Anda dapat menggunakan salah satu metode berikut untuk menggunakan konektor:

Baca file dependensi tambahan dari OSS

Pekerjaan JAR Flink tidak mendukung pembacaan file konfigurasi lokal dari dalam fungsi `main`. Sebagai gantinya, Anda dapat mengunggah file konfigurasi ke bucket OSS yang terkait dengan ruang kerja Flink Anda. Saat menerapkan pekerjaan, Anda kemudian dapat membaca file tersebut dengan menambahkannya sebagai dependensi tambahan. Bagian berikut memberikan contohnya.

1. Buat file konfigurasi bernama `config.properties` untuk menghindari penggunaan password teks biasa dalam kode Anda.

   # 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. Gunakan kode dalam pekerjaan JAR untuk membaca file `config.properties` yang disimpan di bucket OSS.

   Metode 1: Baca dari Bucket OSS yang dilampirkan ke ruang kerja

   1. Di Konsol pengembangan Realtime Compute for Apache Flink, buka halaman Resource Management di panel navigasi sebelah kiri dan unggah file tersebut.

   2. Saat pekerjaan berjalan, file dependensi tambahan yang Anda tambahkan selama penerapan dimuat ke direktori /flink/usrlib pod tempat pekerjaan berjalan.

   3. Kode berikut menunjukkan contoh cara membaca file konfigurasi.

                  Properties properties = new Properties();
                  Map<String,String> configMap = new HashMap<>();
      
                  try (InputStream input = new FileInputStream("/flink/usrlib/config.properties")) {
                      // Load the properties file.
                      properties.load(input);
                      // Get property values.
                      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();
                  }

   Metode 2: Baca dari Bucket OSS yang memiliki izin akses oleh ruang kerja

   1. Unggah file konfigurasi ke bucket OSS tujuan.

   2. Anda dapat menggunakan `OSSClient` untuk langsung membaca file yang disimpan di OSS. Untuk informasi selengkapnya, lihat Stream dan Manage access credentials. Berikut adalah contoh kodenya.

      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()))) {
          // read file and process ...
      } finally {
          if (ossClient != null) {
              ossClient.shutdown();
          }
      }

Tulis kode bisnis

1. Kode berikut menunjukkan cara mengintegrasikan sumber data eksternal ke dalam program aliran data Flink. watermark adalah mekanisme dalam Flink yang mengukur kemajuan dalam event time dan sering digunakan bersama timestamp. Contoh ini tidak menggunakan kebijakan watermark. Untuk informasi selengkapnya, lihat Watermark Strategies.

            // Integrate an external data source into the Flink data stream program.
           // WatermarkStrategy.noWatermarks() indicates that no watermark policy is used.
           DataStreamSource<String> stream = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka Source");

2. Kode berikut menunjukkan transformasi operator. Contoh ini mengonversi DataStream<String> menjadi DataStream<Student>. Untuk informasi selengkapnya tentang transformasi operator kompleks dan metode pemrosesan, lihat Flink Operators.

             // An operator that transforms the data structure to student.
             DataStream<student> source = stream
                   .map(new MapFunction<String, student>() {
                       @Override
                       public student map(String s) throws Exception {
                           // The data is separated by commas.
                           String[] data = s.split(",");
                           return new student(Integer.parseInt(data[0]), data[1], Integer.parseInt(data[2]));
                       }
                   }).filter(student -> student.score >=60); // Filter data to find scores greater than or equal to 60.

Kemas pekerjaan

Anda dapat mengemas pekerjaan menggunakan plugin `maven-shade-plugin`.

<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>

            <!-- We use the maven-shade-plugin to create a fat jar that contains all required dependencies. -->
            <!-- Modify the value of <mainClass> if your program entry point changes. -->
            <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>
                        <!-- Remove some unnecessary dependencies. -->
                        <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>
                                    <!-- Do not copy the signatures in the META-INF folder. Otherwise, this may cause a security exception when you use the JAR file. -->
                                    <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>