Realtime Compute for Apache Flink:Develop JAR jobs

Configure Flink environment dependencies

The following code shows some basic Flink dependencies. You may also need to add dependencies for log files. For a complete list of dependencies, see Complete sample code at the end of this topic.

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

Connector dependencies and usage

To read and write data using DataStream, you must use the corresponding DataStream connector. The Maven central repository contains VVR DataStream connectors that you can use directly during job development.

You can use one of the following methods to use a connector:

Read additional dependency files from OSS

Flink JAR jobs do not support reading local configuration files from within the `main` function. Instead, you can upload the configuration file to an OSS bucket associated with your Flink workspace. When you deploy the job, you can then read the file by adding it as an additional dependency. The following sections provide examples.

1. Create a configuration file named `config.properties` to avoid using plaintext passwords in your code.

   # 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. Use code in the JAR job to read the `config.properties` file stored in the OSS bucket.

   Method 1: Read from the OSS Bucket that is attached to the workspace

   1. In the Realtime Compute for Apache Flink development console, go to the Resource Management page in the navigation pane on the left and upload the file.

   2. When the job runs, the additional dependency file that you added during deployment is loaded into the /flink/usrlib directory of the pod where the job runs.

   3. The following code shows an example of how to read the configuration file.

                  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();
                  }

   Method 2: Read from an OSS Bucket that the workspace has permission to access

   1. Upload the configuration file to the destination OSS bucket.

   2. You can use `OSSClient` to directly read the file stored in OSS. For more information, see Stream and Manage access credentials. The following is a sample code.

      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();
          }
      }

Write business code

1. The following code shows how to integrate an external data source into the Flink data stream program. A watermark is a mechanism in Flink that measures progress in event time and is often used with timestamps. This example does not use a watermark policy. For more information, see 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. The following code shows an operator transformation. The example converts DataStream<String> to DataStream<Student>. For more information about complex operator transformations and processing methods, see 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.

Package the job

You can package the job using the `maven-shade-plugin` plug-in.

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