> For the complete documentation index, see [llms.txt](https://eric-lo.gitbook.io/flink-lab/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://eric-lo.gitbook.io/flink-lab/prepare-your-flink-java-program.md).

# Prepare your Flink Java program

SSH to node-1 to clone the git repo for this lab to your home directory. You can do this in your computer, but node-1 provides the necessary Java environment to compile the program.

```bash
ssh node-1.cse.cuhk.edu.hk
git clone https://github.com/hack3ric/CUHK-CSE-Lab-Flink.git
cd CUHK-CSE-Lab-Flink
```

Open the file `src/main/java/cslab/flinklab/FlinkExample.java`

*Line 28*: Define the stream source as our custom class CustomSource.

*Lines 31-36*: Do the transformation.\
\- *Line 33*: **flatMap**(*map\_function*);\
     **flatMap()** emits 0 or more records (i.e., 1:n mapping) whereas **map()** emits 1 record (i.e., 1:1 mapping)\
     the *map\_function* this time is GetWordCount().\
&#x20;    For each sentence, emit multiple records \<word1, 1>, \<word2, 1>.\
\- *Line 35*: keyBy(0);  //group by 1st element.\
\- *Line 36*: sum(1);  //aggregate each group by 2nd element.

*Lines 39-40*: Define a PrintSink.

```java
package cslab.flinklab;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.util.Collector;

public class FlinkExample {

    private static final Logger LOG = LoggerFactory.getLogger(FlinkExample.class);

    public static void main(String[] args) throws Exception {
        // checking input parameters
        final ParameterTool params = ParameterTool.fromArgs(args);

        // set up the execution environment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.getConfig().setGlobalJobParameters(params);
        env.setParallelism(params.getInt("parallelism", 1));

        // define the Source
        DataStream<String> streamSource = env.addSource(new CustomSource());

        // transform
        DataStream<Tuple2<String, Integer>> words = streamSource
                // splitting sentences to (word, 1)
                .flatMap(new GetWordCount())
                // group by words and sum their occurrences
                .keyBy(0)
                .sum(1);

        // emit result to the standard output
        SinkFunction<Tuple2<String, Integer>> printSink = new PrintSinkFunction<>();
        words.addSink(printSink).name("StdoutSink");

        // execute program
        env.execute("Flink Streaming Example");
    }

    // *************************************************************************
    // Custom FlatMap Function
    // *************************************************************************
    public static class GetWordCount implements FlatMapFunction<String, Tuple2<String, Integer>> {

        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
            // split the sentence into words
            String[] words = value.toLowerCase().split("\\W+");

            // emit each word with a count of 1
            for (String word : words) {
                if (word.length() > 0) {
                    out.collect(new Tuple2<>(word, 1));
                }
            }
        }
    }
}
```

If you are interested, also have a look at the CustomSource class.