Realtime Compute for Apache Flink:Develop a JAR draft

Configure environment dependencies for Apache Flink

The following sample code provides an example of Flink-related dependencies. You may also need to add some dependencies related to log files. For more information about the complete dependencies, see the "Complete sample code" section of this topic.

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

Use connector dependencies

If you want to read and write data in DataStream mode, you must use the related DataStream connector to connect to Realtime Compute for Apache Flink. You can use the VVR DataStream connectors that are stored in the Maven central repository to develop drafts.

You can use connectors in one of the following ways:

Read data from a dependency file stored in OSS

You cannot read data from a local configuration file by using the Main function for a JAR deployment for Realtime Compute for Apache Flink. In this case, you can upload the configuration file to an Object Storage Service (OSS) bucket and add the configuration file to the Additional Dependencies field in the Create Jar Deployment dialog box when you create a JAR deployment to read data from the configuration file. Example:

1. Create the configuration file config.properties and do not use plaintext 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. Run code in a JAR draft to read data from the configuration file config.properties that is stored in an OSS bucket.

   Method 1: Read data from the configuration file in the OSS bucket that is associated with your workspace

   1. In the left-side navigation pane of the development console of Realtime Compute for Apache Flink, click Artifacts. On the Artifacts page, upload the file.

   2. When the deployment is running, the additional dependency file that is added to the Additional Dependencies field is loaded to the /flink/usrlib directory of the pod in which the deployment is running.

   3. Run the following code in the JAR draft to read data from the configuration file:

                  Properties properties = new Properties();
                  Map<String,String> configMap = new HashMap<>();
      
                  try (InputStream input = new FileInputStream("/flink/usrlib/config.properties")) {
                      // Load the property file.
                      properties.load(input);
                      // Obtain the 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 data from the configuration file in the OSS bucket on which the workspace of Realtime Compute for Apache Flink has access permissions

   1. Upload the configuration file config.properties to the OSS bucket.

   2. Use OSSClient in a JAR draft to read data from the configuration file in the OSS bucket. For more information, see Streaming download and Manage access credentials. 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. Integrate external data sources into data streaming programs of Realtime Compute for Apache Flink. Watermarks are computing strategies based on time semantics in Realtime Compute for Apache Flink. In most cases, watermarks are used together with timestamps. In this example, watermark strategies are not used. For more information, see Generating Watermarks.

            // Integrate external data sources into data streaming programs of Realtime Compute for Apache Flink.
           // Specify WatermarkStrategy.noWatermarks(), which indicates that no watermark strategy is used.
           DataStreamSource<String> stream = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka Source");

2. Convert operators. In this example, the DataStream<String> operator is converted into DataStream<Student>. For more information about how to convert and process more complex operators, see Operators.

             // Convert the operator whose data structure is student.
             DataStream<student> source = stream
                   .map(new MapFunction<String, student>() {
                       @Override
                       public student map(String s) throws Exception {
                           // Separate data with commas (,).
                           String[] data = s.split(",");
                           return new student(Integer.parseInt(data[0]), data[1], Integer.parseInt(data[2]));
                       }
                   }).filter(student -> student.score >=60); // Obtain data records whose score is greater than 60.

Package the JAR file of the draft

Package the JAR file of the draft by using 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>

            <!-- Use maven-shade-plugin to create a fat JAR file that contains all the necessary 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 signature in the META-INF folder. Otherwise, a security error may occur 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>