Develop a JAR job

Configure Maven (optional)

Configure the Maven settings.xml file. If you experience slow speeds or failures when downloading from the Maven central repository, you can switch to the Alibaba Cloud Maven mirror.

  <mirror>
      <id>aliyunmaven</id>
        <mirrorOf>central</mirrorOf>
        <name>Aliyun Maven</name>
        <url>https://maven.aliyun.com/repository/public</url>
    </mirror>

Configure Flink environment dependencies

The following are basic Flink dependencies. You may also need to add logging dependencies. For a complete list of dependencies, see the Complete code example at the end of this topic.

         <!-- Apache Flink dependencies -->
        <!-- These dependencies are set to '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 with the DataStream API, use a DataStream connector to connect to Realtime Compute for Apache Flink. The Maven Central Repository provides VVR DataStream connectors that you can use during development.

You can use a connector in one of the following ways:

Read additional dependencies from OSS

Flink JAR jobs cannot read local configuration files from the main method. Instead, upload the configuration file to your workspace's OSS bucket, add it as an additional dependency during deployment, and read it at runtime. The following section provides an example.

1. Create a configuration file named config.properties to avoid hardcoding credentials 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. In your JAR job, use code to read the config.properties file stored in the OSS bucket.

   Method 1: Read from workspace bucket

   1. In the left-side navigation pane of the Realtime Compute for Apache Flink console, go to the Artifacts page and upload the file.

   2. At runtime, files added in the Additional Dependencies field are loaded into the /flink/usrlib directory of the pod where the job runs.

   3. The following code provides an example of how to read this 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 from authorized bucket

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

   2. Use OSSClient to read the file directly from OSS. For more information, see Streaming download and Manage access credentials. The following code provides an example.

      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 the business logic

1. You can integrate external data sources into Flink data stream programs. A Watermark is a Flink computation strategy that is based on time semantics and is often used with timestamps. Therefore, this example does not use a watermark strategy. For more information, see Watermark strategy.

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

2. Operator transformation converts a DataStream<String> to a DataStream<Student> in this example. For more complex operator transformations and processing methods, see Flink Operators.

             // Operator that converts the data structure to Student.
             DataStream<Student> source = stream
                   .map(new MapFunction<String, Student>() {
                       @Override
                       public Student map(String s) throws Exception {
                           // 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 for data where the score is 60 or higher.

Package the job

Package the job by using the 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 necessary dependencies. -->
            <!-- Modify the value of <mainClass> if your program's 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>
                        <!-- Exclude 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 from the META-INF directory.
                                    Otherwise, security exceptions may be thrown 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>