Develop a Flink application

Objectives: Develop a simple Flink application.

Complexity: Beginner

Time to complete: 30 min

Source code: Download

This tutorial demonstrates how to develop, debug, and run a simple Flink application that reads data from the stream layer, and logs all this data using log4j.

The tutorial covers the following topics:

Set up the Maven project

Download the source code at the beginning of the tutorial and put it in a folder of your choice, or create a folder structure for your project from scratch:

develop-flink-application
└── src
    └── main
        ├── java
        └── resources
        └── scala

You can do this with a single bash command:

mkdir -p develop-flink-application/src/main/{java,resources,scala}

The Maven POM file is similar to the one in the Verify Maven Settings example, however, with an updated parent POM and dependencies sections:

The Parent POM is sdk-stream-bom_${scala.compat.version}, because we need to use Flink-related libraries.

<parent>
    <groupId>com.here.platform</groupId>
    <artifactId>sdk-stream-bom_2.12</artifactId>
    <version>2.42.4</version>
    <relativePath/>
</parent>

The following dependencies are used:

  • com.here.platform.data.client:local-support_${scala.compat.version} to read data from a local data catalog.
  • org.apache.flink:flink-clients_${scala.compat.version} to provide ExecutorFactory for the Flink application.
  • org.apache.flink:flink-streaming-java_${scala.compat.version} to run a Java Flink application.
  • org.apache.flink:flink-streaming-scala_${scala.compat.version} to run a Scala Flink application.
  • com.here.platform.data.client:flink-support_${scala.compat.version} to read data from the data catalogs on the platform.
  • org.slf4j:slf4j-log4j12 to log application results to the console and to the Splunk on the platform.

Dependencies:

<dependencies>
    <dependency>
        <groupId>com.here.platform.data.client</groupId>
        <artifactId>flink-support_${scala.compat.version}</artifactId>
    </dependency>
    <dependency>
        <groupId>com.here.platform.data.client</groupId>
        <artifactId>local-support_2.12</artifactId>
        <exclusions>
            <exclusion>
                <groupId>com.here.platform.data.client</groupId>
                <artifactId>client-core_2.12</artifactId>
            </exclusion>
        </exclusions>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients_${scala.compat.version}</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_${scala.compat.version}</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-scala_${scala.compat.version}</artifactId>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
    </dependency>
</dependencies>

Once you have added all the necessary dependencies to the pom.xml file, the next step would be to write the code of the application and run it.

Write the source code

As mentioned before, the tutorial shows how to write a simple Flink application that reads data from a stream layer and outputs the data in the console. The data is added to the stream layer during the execution of your stream application using OLP CLI. All data read from the layer is logged to the console at the same time using log4j. The configuration of log4j is located in the src/main/resources/log4j.properties file.

Let's look at the implementation of this Flink application. In the code snippet below, you can see that the FlinkQueryApi is used to create a subscription to receive new data added to the stream layer. Once the subscription is created and the DataStreamSource source function that emits new data from stream layer is declared, we map partitions using the SampleMaper class that extends RichMapFunction, which provides setup and teardown methods. The SampleMapper class downloads partitions and logs as a human-readable string.

Java
Scala

/*
 * Copyright (c) 2018-2022 HERE Europe B.V.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import com.here.hrn.HRN;
import com.here.platform.data.client.flink.javadsl.FlinkDataClient;
import com.here.platform.data.client.flink.javadsl.FlinkQueryApi;
import com.here.platform.data.client.flink.javadsl.FlinkReadEngine;
import com.here.platform.data.client.javadsl.Partition;
import com.here.platform.data.client.settings.ConsumerSettings;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class DevelopFlinkApplication {

  private static final Logger LOGGER = LoggerFactory.getLogger(DevelopFlinkApplication.class);

  private static final HRN CATALOG_HRN = HRN.fromString("{{YOUR_CATALOG_HRN}}");
  private static final String STREAMING_LAYER = "streaming-layer";
  private static final String CONSUMER_GROUP_NAME = "flink";

  public static void main(String[] args) throws Exception {

    // Create queryApi for the source catalog
    final FlinkDataClient flinkDataClient = new FlinkDataClient();
    FlinkQueryApi queryApi = flinkDataClient.queryApi(CATALOG_HRN);

    // Create the context in which a streaming program is executed
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    // Specify Kafka consumer group name
    ConsumerSettings consumerSettings =
        new ConsumerSettings.Builder()
            .withGroupName(CONSUMER_GROUP_NAME)
            .withLatestOffset()
            .build();

    // Subscription to receive new data added to the stream layer
    SourceFunction<Partition> subscriptionFunction =
        queryApi.subscribe(STREAMING_LAYER, consumerSettings);

    // Source function that emits new data from stream layer
    DataStreamSource<Partition> partitions = env.addSource(subscriptionFunction);

    // Read partition bytes, map to human-readable string
    // using the Sample Mapper class that implements RichMapFunction recommended to use by Data
    // Client Library,
    // and log using log4j
    partitions.map(new SampleMapper());

    // Trigger the program execution
    env.execute();
  }

  private static class SampleMapper extends RichMapFunction<Partition, String> {
    private transient FlinkDataClient flinkDataClient;
    private transient FlinkReadEngine flinkReadEngine;

    @Override
    public void open(Configuration parameters) {
      // Create read engine to read the partition bytes
      flinkDataClient = new FlinkDataClient();
      flinkReadEngine = flinkDataClient.readEngine(CATALOG_HRN);
    }

    @Override
    public String map(Partition partition) {
      String partitionContent = new String(flinkReadEngine.getDataAsBytes(partition));
      LOGGER.info(partitionContent);
      return partitionContent;
    }

    @Override
    public void close() {
      // FlinkDataClient is a heavyweight object that needs to be created once, reused and
      // terminated.
      flinkDataClient.terminate();
    }
  }
}

 

/*
 * Copyright (c) 2018-2022 HERE Europe B.V.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import com.here.hrn.HRN
import com.here.platform.data.client.flink.javadsl.{FlinkDataClient, FlinkReadEngine}
import com.here.platform.data.client.javadsl.Partition
import com.here.platform.data.client.settings.{ConsumerSettings, LatestOffset}
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.slf4j.LoggerFactory

import java.io.Serializable

object DevelopFlinkApplicationScala {

  private final val Logger = LoggerFactory.getLogger(classOf[DevelopFlinkApplication])

  private final val CatalogHRN = HRN.fromString("{{YOUR_CATALOG_HRN}}")
  private final val StreamingLayer = "streaming-layer"
  private final val ConsumerGroupName = "flink"

  def main(args: Array[String]): Unit = {

    // Create queryApi for the source catalog
    val client = new FlinkDataClient
    val queryApi = client.queryApi(CatalogHRN)

    // Create the context in which a streaming program is executed
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // Specify Kafka consumer group name
    val consumerSettings = ConsumerSettings(ConsumerGroupName, LatestOffset)

    // Subscription to receive new data added to the stream layer
    val subscriptionFunction = queryApi.subscribe(StreamingLayer, consumerSettings)

    // Source function that emits new data from stream layer
    val partitions = env.addSource(subscriptionFunction)

    // Read partition bytes, map to human-readable string
    // using the Sample Mapper class that implements RichMapFunction recommended to use by Data Client Library,
    // and log using log4j
    partitions.map(new SampleMapper(CatalogHRN))

    // Trigger the program execution
    env.execute

  }

  class SampleMapper(hrn: HRN) extends RichMapFunction[Partition, String] with Serializable {
    @transient
    private lazy val flinkDataClient: FlinkDataClient = new FlinkDataClient()

    @transient
    private lazy val flinkReadEngine: FlinkReadEngine =
      // Create read engine to read the partition bytes
      flinkDataClient.readEngine(hrn)

    override def map(partition: Partition): String = {
      val partitionContent = new String(flinkReadEngine.getDataAsBytes(partition))
      Logger.info(partitionContent)
      partitionContent
    }

    override def close(): Unit =
      // FlinkDataClient is a heavyweight object that needs to be created once, reused and terminated.
      flinkDataClient.terminate()
  }
}

 

Once the code is written, you can prepare resources and run the application.

Run the application

To run the application, you need to prepare the resources - create a catalog with a stream layer where the data will be written.

In this tutorial we will run our application locally, therefore, it will be enough for us to create a local catalog. No authentication and no access to the external network are needed to run this tutorial as we use local catalogs. Since they are contained in a local machine, they are not subject to naming conflicts within your realm and you can use any name you want.

To create a local catalog with a stream layer, you will need the catalog-with-stream-layer.json config file that contains configuration allowing you to create a catalog with a layer using only one OLP CLI command:

olp local catalog create streaming-catalog streaming-catalog --config catalog-with-stream-layer.json

Once the catalog is created, you need to replace the placeholder in the source code with the catalog HRN from the previous command response. Replace placeholders in the following Java/Scala classes:

Java
Scala
 
private static final HRN catalogHRN = HRN.fromString("{{YOUR_CATALOG_HRN}}"); 
 

private val catalogHRN = HRN("{{YOUR_CATALOG_HRN}}")

After you have replaced the placeholder, run the application from the root of the downloaded tutorial using the following command:

Java
Scala
 
mvn compile exec:java -D"exec.mainClass"="DevelopFlinkApplication" -Dhere.platform.data-client.endpoint-locator.discovery-service-env=local 
 
 mvn compile exec:java -D"exec.mainClass"="DevelopFlinkApplicationScala" -Dhere.platform.data-client.endpoint-locator.discovery-service-env=local 
 

The command has the following parameters:

  • exec.mainClass – entry point to run your application.
  • here.platform.data-client.endpoint-locator.discovery-service-env=local – configures the Data Client Library to only use local catalogs.

At this moment, the application is running, but there is no data in the input catalog. Let's put some partitions in it. Open a new console window and run the following bash script that uploads data with the First HERE Platform Flink Application content to the stream layer 10 times every 5 seconds:

Bash

#!/usr/bin/env bash

# Copyright (c) 2018-2022 HERE Europe B.V.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.



# Local catalog hrn with stream layer
CATALOG_HRN=$1
# Folder with data that is uploaded to the stream layer
FOLDER_WITH_DATA="./src/main/resources/data"


# Upload data to the stream layer 10 times every 5 seconds
for i in {1..10}
do
  olp local catalog layer stream put ${CATALOG_HRN} streaming-layer --input "${FOLDER_WITH_DATA}"
  sleep 5
done
 

To run the script, execute the following command:

bash scripts/populate-streaming-data.sh {{YOUR_CATALOG_HRN}}

Once the script is running, you should see the logs with the same information in the console of the running application.

Attach the debugger

In this chapter you will learn how to debug Flink applications using the capabilities of Intellij Idea, or rather you will learn how to attach to the process to start debugging if you run your program through the console.

In order to configure the debugger, you need to set up the MAVEN_OPTS variable. To do so, stop the running application and run the following command:

JDK 5-8
Java 9 or later
 
export MAVEN_OPTS=-agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=5005 

 
export MAVEN_OPTS=-agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=*:5005 
 

The option has the following parameters:

  • address – the port that will be used for debugging. The tutorial uses port 5005, however, you can use any free port.
  • server=y – specifies that the process should listen for incoming debugger connections (act as a server).
  • suspend=n – specifies that the process should wait until the debugger has been connected.

Now you can run your application using the following command:

Java
Scala
 mvn compile exec:java -D"exec.mainClass"="DevelopFlinkApplication"
-Dhere.platform.data-client.endpoint-locator.discovery-service-env=local  
 mvn compile exec:java -D"exec.mainClass"="DevelopFlinkApplicationScala"
-Dhere.platform.data-client.endpoint-locator.discovery-service-env=local  

Make sure that line Listening for transport dt_socket at address: 5005 appears in the logs.

Let's set the breakpoint before we attach to the running application, for example, in the line where the partitions are mapped. From this line, you can get a lot of useful information about the partition you downloaded:

Now we can attach to the process using Run > Attach to Process and select the process with the specified 5050 port.

Once the process is attached and your program is running and waiting for the data to appear in the stream layer, you need to run the script again to inject data:

bash scripts/populate-streaming-data.sh {{YOUR_CATALOG_HRN}}

After running the scripts/populate-streaming-data.sh script, the debugger should stop at the breakpoint as soon as new data appears in the catalog and your application will start reading it. Now you can step through the code and inspect the content of the variables and the stack trace.

You can also use the standart Java Debugger instead of the debugger in the Intellij IDEA.

Conclusion

In this tutorial, you have learned the stages of the Flink program development. To learn how to run a Flink application on the platform and to get acquainted with such monitoring tools as Splunk, Grafana, Flink UI, and Platform Billing Page, see the Run a Flink application on the platform tutorial.

Further information

For more details on the topics covered in this tutorial, see the following sources:

results matching ""

    No results matching ""