If you remember in mv’ing batch flink to streaming (easy breezy), we discussed how Flink has the DataSet API for batch and the DataStream API for streaming. The Flink team has been talking for some time about treating batch as a “special case” of streaming and finally in version 1.12 they have “soft-deprecated” the DataSet API.

We are here now and have a BATCH Execution Mode! What do we have to do? Basically the same work we had to do moving off DataSet API to DataStream API. For our (twisted) example, we can convert our streaming app back to batch with changing one line of code. Heck, only change a half of line of code.

Let’s grab the line starting with…

DataStream lines = …

Swap out…

env.socketTextStream("localhost", 9999);

With…

env.readTextFile(params.get("input"));

ALMOST DONE! We need configure the job to run in batch mode. We could add the following line just after our instantiation of StreamExecutionEnvironment.

env.setRuntimeMode(RuntimeExecutionMode.BATCH);

That said, the recommendation from the documentation is to use this command-line switch instead when submitting your job.

-Dexecution.runtime-mode=BATCH

So let’s run our job like we did before in hello world with flink (from scratch).

MBP15:bin lmartin$ pwd
/Users/lmartin/blog/flink-1.12.2/bin
MBP15:bin lmartin$ ./flink run --class wordcount.WordCountBatchWithDataStream \
>  -Dexecution.runtime-mode=BATCH \
>  ~/blog/flink-exploration/target/flink-exploration-0.0.1-SNAPSHOT.jar \
>  --input file:///Users/lmartin/blog/flink-exploration/src/test/resources/BitOfGreenEggsAndHam.txt 
Job has been submitted with JobID df8565fff40cc00356d376920249ca6c
Program execution finished
Job with JobID df8565fff40cc00356d376920249ca6c has finished.
Job Runtime: 858 ms

MBP15:bin lmartin$ 

As I left the sink setup like that of the streaming solution, here is the “Stdout” tab with results again.

In summary, Flink has melded the DataSet API into the DataStream API and likely will cause less confusion to most developers going forward. If you need it, here’s the source code.

This is the follow-up post mentioned in hello world with flink (from scratch) that shows how to take that simple batch application and modify it to be a real-time streaming job.

Create a Source

Streaming apps need a continuous source of data to process. This could be all kinds of things including using Apache Kafka, but for our example we are continuing to use the KISS principle. We will just connect to a socket and expect strings of words to be sent to us. To stand up a test harness (manual) generator, just run the following command.

nc -l 9999

This will just put the cursor on the following line and you can type data and hit the <ENTER> key to submit that string out over port 9999.

NOTE: You might simply <CTRL-Z> this until you are ready to submit your job.

Modify the Code

If you just completed the code in hello world with flink, then you are ready to go. If not, you can clone the blog-wc-batch branch of my flink-exploration github project.

MBP15:blog2 lmartin$ pwd
/Users/lmartin/blog2
MBP15:blog2 lmartin$ git clone --branch blog-wc-batch --single-branch https://github.com/lestermartin/flink-exploration.git
MBP15:blog2 lmartin$ tree
.
└── flink-exploration
    ├── README.md
    ├── pom.xml
    └── src
        ├── main
        │   └── java
        │       └── wordcount
        │           └── WordCountBatch.java
        └── test
            └── resources
                ├── BitOfGreenEggsAndHam.txt
                └── GreenEggsAndHam.txt

7 directories, 5 files
MBP15:blog2 lmartin$ 

NOTE: You could also just pull down the blog-wc-streaming branch if you don’t want to make the changes yourself. 😉

Step 1: Duplicate the Batch Class

Make a copy of WordCountBatch.java and call it WordCountStreaming.java.

Step 2: Update the Imports

You could do this while modifying the new streaming class, but I’m just trying to help you out here!! Add the following two lines.

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

Eventually, you can safely remove the DataSet and ExecutionEnvironment statements.

Step 3: Adjust the Boilerplating

Update the first line of the boilerplate code to use the new “environment” class just imported so it looks like the following.

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

Step 4: Stream Instead of Read

Change this line that was reading a file.

// read the input file
DataSet<String> lines =
  env.readTextFile(params.get("input"));

To now listen to the socket for continuous feeds of data.

// listen to a socket
DataStream<String> lines = 
  env.socketTextStream("localhost", 9999);

Notice that we not only called a different method on the env object, we are now creating a DataStream instead of a DataSet.

Step 5: Adapt to DataStream

The wordcounts streaming collection should now be of type DataStream and the groupBy() method after the flatMap() needs to be changed to a keyBy(). This code block should look similar to the following.

        //calculate word counts
        DataStream<Tuple2<String, Integer>> wordcounts =
                // split up the lines in pairs (2-tuples) containing: (word,1)
                lines.flatMap(new Tokenizer())
                        // group by the tuple field "0" and sum up tuple field "1"
                        .keyBy(0).sum(1);

Step 6: Simplify the Output

In a “real” application, we’ll likely need a sink that persists some/all of the streaming data somewhere, but let’s replace the whole // output the dataset section with just these two lines.

        wordcounts.print();
        env.execute("WordCount Streaming Example");

This will simply send our results to “standard out” which we can view in the Flink UI once we start our continuous job running.

Here is the diff output from my IntelliJ setup.

Here is a CLI diff output of the same changes.

MBP15:wordcount lmartin$ pwd
/Users/lmartin/blog/flink-exploration/src/main/java/wordcount
MBP15:wordcount lmartin$ ls
WordCountBatch.java	WordCountStreaming.java
MBP15:wordcount lmartin$ diff WordCountBatch.java WordCountStreaming.java 
4,5d3
< import org.apache.flink.api.java.DataSet;
< import org.apache.flink.api.java.ExecutionEnvironment;
7a6,7
> import org.apache.flink.streaming.api.datastream.DataStream;
> import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
10c10
< public class WordCountBatch
---
> public class WordCountStreaming
16c16
<         ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
---
>         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
20,21c20,21
<         // read the input file
<         DataSet<String> lines = env.readTextFile(params.get("input"));
---
>         // listen to a socket
>         DataStream<String> lines = env.socketTextStream("localhost", 9999);
24c24
<         DataSet<Tuple2<String, Integer>> wordcounts =
---
>         DataStream<Tuple2<String, Integer>> wordcounts =
28c28
<                         .groupBy(0).sum(1);
---
>                         .keyBy(0).sum(1);
30,36c30,31
<         // output the dataset
<         if (params.has("output"))
<         {
<             wordcounts.writeAsCsv(params.get("output"), "\n", " ");
<             // the next line triggers the lazy execution to begin
<             env.execute("WordCount Batch Example");
<         }
---
>         wordcounts.print();
>         env.execute("WordCount Streaming Example");
MBP15:wordcount lmartin$ 

Build and Submit

Use maven to package up a jar file again. Our command-line submission doesn’t need any parameters to be sent into the class so it could be much more succinct.

MBP15:bin lmartin$ ./flink run \
> --class wordcount.WordCountStreaming \
> ~/blog/flink-exploration/target/flink-exploration-0.0.1-SNAPSHOT.jar 
Job has been submitted with JobID 851364dd2d8197f3d717e9349437ff7e

We should be able to see the running job in the http://localhost:8081 UI.

As an alternative, we could also set up the jar to be used for job submission from the UI. There is a “Submit New Job” link in the left nav and the rendered page has a “+ Add New” button in the upper-right of the page. Once you select your jar file from the upload dialogue box you can click on the name of the jar file in the UI to see some additional submit options.

For our simple job, we only need to update the “Entry Class” box with wordcount.WordCountStreaming and push the “Submit” button. This should send us straight to the “Overview” tab of this newly launched job.

Test It Out

One way or another, we should have the job running now. If it failed, go back and make sure the nc command is still working. If it is then you should be able to add some lines of text such as the following.

MBP15:~ lmartin$ nc -l 9999
now is the time for all good programmers to learn flink
flink is now all the time and programmers love to learn

While we can dive through the running job’s UI pages to find the “standard out” data, in this simple setup it will be much easier to just click on “Task Managers” on the left nav.

Click on the link just below the “Path, ID” heading and on the newly rendered page and then select the “Stdout” tab to see what is being counted by the job.

Looks about right as we have 2 instances of both programmers and learn showing up.

Obviously, we are barely scratching the surface of Flink. Fortunately, moving a batch application to streaming will keep the bulk of the core code the same with the most updates focused on adapting the source for sure, and possibly the sink.

Tear It Down!

Click on the “Cancel Job” link in the upper-right corner of the job detail page.

Perform a <CTRL-Z> on the nc command.

MBP15:~ lmartin$ nc -l 9999
now is the time for all good programmers to learn flink
flink is now all the time and programmers love to learn
^Z
[5]+  Stopped                 nc -l 9999
MBP15:~ lmartin$ 

You could always shut down Flink on your laptop if you’re done.

MBP15:bin lmartin$ pwd
/Users/lmartin/blog/flink-1.12.2/bin
lmartin-MBP15-8694:bin lmartin$ ./stop-cluster.sh 
Stopping taskexecutor daemon (pid: 76383) on host lmartin-MBP15-8694.local.
Stopping standalonesession daemon (pid: 76140) on host lmartin-MBP15-8694.local.
MBP15:bin lmartin$

If you have heard about Apache Flink, but have not gotten your hands dirty yet then this is the blog post for you. We will stand up a local development environment, write some simple code (much more to Flink than we will get into during this post), and then test it out.

And… yes, like my functional programming and big data (what a pair) and big data api’s look a lot alike (code comparisons with flink, kafka, spark, trident and pig) we’ll use the darn word count example. Also, it won’t be as robust as the one from project’s source code.

Setup Environment

<shamelessProductPlug>
  If you already have a Flink runtime environment, 
  such as Cloudera Data Platform (CDP),
  then you are ALREADY there!!
</shamelessProductPlug>

If not, or if you just want to set something small up on your workstation, let’s keep it simple (KISS) and just go over to Flink’s download page and select an appropriate version. Most often the latest version is the best and for me at time of this article, I went with 1.12.2 — specifically I downloaded Apache Flink 1.12.2 for Scala 2.11. Then we just need to unwind it and start it up.

MBP15:blog lmartin$ pwd
/Users/lmartin/blog
MBP15:blog lmartin$ ls
flink-1.12.2-bin-scala_2.11.tgz
MBP15:blog lmartin$ tar -xvf flink-1.12.2-bin-scala_2.11.tgz 
    ... lines removed ...
MBP15:blog lmartin$ tree -L 2
.
├── flink-1.12.2
│   ├── LICENSE
│   ├── NOTICE
│   ├── README.txt
│   ├── bin
│   ├── conf
│   ├── examples
│   ├── lib
│   ├── licenses
│   ├── log
│   ├── opt
│   └── plugins
└── flink-1.12.2-bin-scala_2.11.tgz

9 directories, 4 files
MBP15:blog lmartin$ cd flink-1.12.2/bin
MBP15:bin lmartin$ ./start-cluster.sh 
Starting cluster.
Starting standalonesession daemon on host MBP15.local.
Starting taskexecutor daemon on host MBP15.local.
MBP15:bin lmartin$ 

Now we can point our browser to http://localhost:8081 to see the Flink UI.

That wasn’t too hard, now was it?

Code It!

In your favorite IDE, create a Maven Java project. Here’s how I identified my project, but call yours whatever you would like.

    <groupId>com.github.lestermartin</groupId>
    <artifactId>flink-exploration</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <packaging>jar</packaging>

And, yes… for those front-row kids, that suggests that all of this code is in github (it is!). The posting will assuming you are building this following my notes, but obviously could just clone or download my blog-wc-batch branch instead.

You will also need some dependencies, so add these to your pom.xml and be mindful of the Flink version.

    <properties>
        <flink.version>1.12.2</flink.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.12</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>
    </dependencies>

Next up, create a wordcount Java package and within that a new Java class named WordCountBatch that has a main() method in it.

NOTE: This is a batch program — I will publish a follow-up blog post of modifying this same app to be a streaming solution. Fortunately, only a few lines of code will change and I thought it would be easier for this first attempt at coding a Flink job if we stuck to that KISS principle we used earlier.

Let’s just blast in some imports that will help us along the way.

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.util.Collector;

Before we do anything else, let’s just make sure we can build our empty class with a mvn clean and install. We can verify the jar file is present, too.

MBP15:target lmartin$ pwd
/Users/lmartin/blog/flink-exploration/target
MBP15:target lmartin$ ls *.jar
flink-exploration-0.0.1-SNAPSHOT.jar
MBP15:target lmartin$ 

At the beginning of our main() method, add in a few lines of boilerplate code to deal with whatever execution environment we find ourselves in during job submission and to be able to access any command-line arguments that were passed in (ex: input file and output location).

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
ParameterTool params = ParameterTool.fromArgs(args);
env.getConfig().setGlobalJobParameters(params);

This line will allow us to read in a text file from a command-line argument.

DataSet<String> lines = env.readTextFile(params.get("input"));

To help us, add the following inner class that can break a space-delimited string into single words and then morph each word into a KVP of (word,1) that we will use in a transformation of the lines dataset in just a minute.

//user defined function
public static final class Tokenizer 
    implements FlatMapFunction<String, Tuple2<String, Integer>> {

    @Override
    public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
        // normalize and split the line
        String[] tokens = value.toLowerCase().split("\\W+");

        // emit the pairs
        for (String token : tokens) {
            if (token.length() > 0) {
                out.collect(new Tuple2<>(token, 1));
            }
        }
    }
}

With the heavy lifting code now available as a user defined function, the rest of the word count processing is easily taken care of from the Flink API itself.

DataSet<Tuple2<String, Integer>> wordcounts =
        // split up the lines in pairs (2-tuples) containing: (word,1)
        lines.flatMap(new Tokenizer())
                // group by the tuple field "0" and sum up tuple field "1"
                .groupBy(0).sum(1);

Lastly, we can write the wordcounts dataset out to disk to review the calculations.

wordcounts.writeAsCsv(params.get("output"), "\n", " ");
// the next line triggers the lazy execution to begin
env.execute("WordCount Batch Example");

Package up the jar file again with the mvn install command.

Run It!

Create a simple text file to test with. In my github project referenced earlier, this is in src/test/resources/BitOfGreenEggsAndHam.txt.

DO WOULD YOU LIKE GREEN EGGS AND HAM
I DO NOT LIKE THEM SAM I AM
I DO NOT LIKE GREEN EGGS AND HAM

Submit the job from the CLI. Be mindful of your classname, jar file location, and the input & output file names to fit your specific setup.

MBP15:bin lmartin$ pwd
/Users/lmartin/blog/flink-1.12.2/bin
MBP15:bin lmartin$ ./flink run --class wordcount.WordCountBatch \
> ~/blog/flink-exploration/target/flink-exploration-0.0.1-SNAPSHOT.jar \
> --input file:///Users/lmartin/blog/flink-exploration/src/test/resources/BitOfGreenEggsAndHam.txt \
> --output file:///Users/lmartin/blog/data/wc-tinyBit-OUT
Job has been submitted with JobID c2079da12c2c0e89a56a1b325f75e5dc
Program execution finished
Job with JobID c2079da12c2c0e89a56a1b325f75e5dc has finished.
Job Runtime: 241 ms

MBP15:bin lmartin$ 

This batch job took less than a second to complete since it was so small. You can find the job in the Flink UI by looking down at the completed jobs.

As expected, the output file is even smaller.

MBP15:data lmartin$ pwd
/Users/lmartin/blog/data
MBP15:data lmartin$ ls
wc-tinyBit-OUT
MBP15:data lmartin$ cat wc-tinyBit-OUT 
am 1
and 2
do 3
eggs 2
green 2
ham 2
i 3
like 3
not 2
sam 1
them 1
would 1
you 1
MBP15:data lmartin$

Remember, you can always pull down the blog-wc-batch branch of flink-exploration github project if that would be easier. There is even a complete text version of Green Eggs and Ham to word count on. If you run it, comment below on the number of times green, eggs, and ham were present. To help make sure it all ran correctly, there were 82 instances of not in the text.