Realtime Compute for Apache Flink:Mengembangkan pekerjaan JAR

Konfigurasi dependensi lingkungan Flink

Berikut adalah dependensi dasar Flink. Anda mungkin juga perlu menambahkan dependensi logging. Untuk daftar lengkap dependensi, lihat Contoh kode lengkap di akhir topik ini.

         <!-- Dependensi Apache Flink -->
        <!-- Dependensi ini diatur ke 'provided' karena tidak boleh dikemas ke dalam file JAR. -->
        <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 dengan API DataStream, gunakan konektor DataStream untuk terhubung ke Realtime Compute for Apache Flink. Repositori Maven Central menyediakan konektor DataStream VVR yang dapat Anda gunakan selama pengembangan.

Anda dapat menggunakan konektor dengan salah satu cara berikut:

Membaca dependensi tambahan dari OSS

Pekerjaan JAR Flink tidak dapat membaca file konfigurasi lokal dari metode main. Sebagai gantinya, unggah file konfigurasi ke bucket OSS ruang kerja Anda, tambahkan sebagai dependensi tambahan selama penerapan, dan baca saat runtime. Bagian berikut memberikan contohnya.

1. Buat file konfigurasi bernama config.properties untuk menghindari hardcoding kredensial 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. Dalam pekerjaan JAR Anda, gunakan kode untuk membaca file config.properties yang disimpan di bucket OSS.

   Metode 1: Baca dari bucket ruang kerja

   1. Di panel navigasi kiri Konsol Realtime Compute for Apache Flink, buka halaman Artifacts dan unggah file tersebut.

   2. Saat runtime, file yang ditambahkan di bidang Additional Dependencies dimuat ke direktori /flink/usrlib pod tempat pekerjaan berjalan.

   3. Kode berikut memberikan contoh cara membaca file konfigurasi ini.

                  Properties properties = new Properties();
                  Map<String,String> configMap = new HashMap<>();
                  try (InputStream input = new FileInputStream("/flink/usrlib/config.properties")) {
                      // Muat file properti.
                      properties.load(input);
                      // Dapatkan nilai properti.
                      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 yang diotorisasi

   1. Unggah file konfigurasi ke bucket OSS target.

   2. Gunakan OSSClient untuk membaca file langsung dari OSS. Untuk informasi selengkapnya, lihat Unduhan streaming dan Mengelola kredensial akses. Kode berikut memberikan contohnya.

      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()))) {
          // baca file dan proses ...
      } finally {
          if (ossClient != null) {
              ossClient.shutdown();
          }
      }

Tulis logika bisnis

1. Anda dapat mengintegrasikan sumber data eksternal ke dalam program aliran data Flink. Watermark adalah strategi komputasi Flink yang berbasis semantik waktu dan sering digunakan bersama timestamp. Oleh karena itu, contoh ini tidak menggunakan strategi watermark. Untuk informasi selengkapnya, lihat Strategi Watermark.

            // Integrasikan sumber data eksternal ke dalam program aliran data Flink.
           // WatermarkStrategy.noWatermarks() menunjukkan bahwa tidak ada strategi watermark yang digunakan.
           DataStreamSource<String> stream = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka Source");

2. Transformasi operator mengonversi DataStream<String> menjadi DataStream<Student> dalam contoh ini. Untuk transformasi operator dan metode pemrosesan yang lebih kompleks, lihat Operator Flink.

             // Operator yang mengonversi struktur data menjadi Student.
             DataStream<Student> source = stream
                   .map(new MapFunction<String, Student>() {
                       @Override
                       public Student map(String s) throws Exception {
                           // Data dipisahkan oleh koma.
                           String[] data = s.split(",");
                           return new Student(Integer.parseInt(data[0]), data[1], Integer.parseInt(data[2]));
                       }
                   }).filter(student -> student.score >=60); // Filter data dengan skor 60 atau lebih tinggi.

Mengemas pekerjaan

Kemas pekerjaan dengan menggunakan 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>
            <!-- Kami menggunakan maven-shade-plugin untuk membuat fat JAR yang berisi semua dependensi yang diperlukan. -->
            <!-- Ubah nilai <mainClass> jika titik masuk program Anda berubah. -->
            <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>
                        <!-- Kecualikan dependensi yang tidak diperlukan. -->
                        <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>
                                    <!-- Jangan salin tanda tangan dari direktori META-INF.
                                    Jika tidak, pengecualian keamanan mungkin terjadi saat Anda menggunakan file 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>