Playing PyFlink in a Nutshell
Tags: bigdata, eventdriven, python, tutorial
Published: 2022-10-24 01:58:01

Last week, we introduced [how to build a PyFlink experiment environment](https://blog.devgenius.io/playing-pyflink-from-scratch-65c18908c366), and today we will use that experimental environment to explore the possibilities of PyFlink.

PyFlink is a general purpose streaming framework and abstracts streaming processing into four levels.

1. SQL
2. Table API
3. DataStream
4. Stateful Stream Processing

The closer to the bottom the more flexibility is available, but also requiring writing more code. I would like to be able to do almost everything with PyFlink, so let's get started with the basic concepts of PyFlink development from a DataStream perspective.

This article will introduce a few key points of PyFlink development with simple descriptions and examples, but will not mention the implementation details of Flink.

## DataStream Concept

The development of DataStream will follow the following process.

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/ptyqms28nxyykvee7qqx.png)

Basically, we get streaming data from a source, process it, and output it to somewhere.

This is expressed in PyFlink as follows.

```python
ds = env.add_source(kafka_consumer)
ds = ds.map(transform, output_type=output_type_info)
ds.add_sink(kafka_producer)
```

Source and sink are easy to understand, but the key is what processing can be used?

In the official document there is a list of all available operations.
https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/datastream/operators/overview/

For the above example, we used `map`.

There is also an example of the operation in the Flink artifact, the code is placed at `./examples/python/datastream/basic_operations.py`。

## What is the difference between `map` and `flat_map`?

There are two similar operations in the operation list, `map` and `flat_map`, what is the difference between these two operations?

The difference is in the number of generated outputs.

In the case of `map`, an input event generates one and only one output event; on the other hand, `flat_map` can generate zero to many output events.

Let's use the actual code as an example.

```python
def map_transform(i: int):
  return i * i
  
def flat_map_transform(i: int):
  for idx in range(i):
    yield idx
    
ds.map(map_transform, output_type=Types.INT())
ds.flat_map(flat_map_transform, output_type=Types.INT())
```

In this example, `map` squares all the input integers and passes them out, one input corresponds to one output. However, `flat_map` outputs a series of events, and the number of output events is determined by the input events.

If the input is `0`, then `yield` of `flat_map` will not be triggered, and nothing will be generated.

## State

State is the best feature of Flink.

Although we have various operations available, many of them actually produce results based on previous events. How do we keep the previous events? This is where `State` comes in.

State can be considered as an internal storage in order to persist data, and the size of State is the summary of every node’s memory.

Nevertheless, State can be persisted in a durable storage like `RocksDB` to gain more scalability.

```python
from pyflink.datastream import StreamExecutionEnvironment, EmbeddedRocksDBStateBackend

env = StreamExecutionEnvironment.get_execution_environment()
env.set_state_backend(EmbeddedRocksDBStateBackend())
```

To use State in Flink framework, there are two key points worth noting.

1. State can only be used in "Keyed Data Stream".
2. State is based on operations and not able to share with others.

All available States and the reference are listing below.
https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/datastream/fault-tolerance/state/

In fact, an example at `./examples/python/datastream/state_access.py` also provides a good demonstration.

## Connect (Shared State)

As mentioned in the previous section, states are operation-based and cannot be shared, but sometimes we indeed need to combine two different streaming states, so what should we do?

Fortunately, Flink provides `connect` to enable us to share the states of different streams within the same job.

By using `connect`, we can combine different streams and use the same operation, so that we can share the same operation state.

To be more concrete, let me provide a practical example. There are two streams.
- Stream 1 provides a mapping between the item identifiers and the item names. When the item name changes, an event `(item_id, item_name)` is sent into the stream, so we just need to save the latest status.
- Stream 2 is the transaction history, including which item was sold and the number of items ordered.

What we want to do is, when any purchase is entered, then we have to sum it up and append the latest item name to it.

This is the classic streaming enrichment pattern, and I explained the enrichment design pattern in detail in [my previous article](https://betterprogramming.pub/design-pattern-of-streaming-enrichment-17a9eb065eca).

Here is the full program example.

```python
import logging, sys
from pyflink.common import WatermarkStrategy, Row
from pyflink.common.serialization import Encoder
from pyflink.common.typeinfo import Types
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors import FileSink, OutputFileConfig, NumberSequenceSource
from pyflink.datastream.execution_mode import RuntimeExecutionMode
from pyflink.datastream.functions import KeyedProcessFunction, RuntimeContext, MapFunction, CoMapFunction, CoFlatMapFunction
from pyflink.datastream.state import MapStateDescriptor, ValueStateDescriptor
from pyflink.common import JsonRowDeserializationSchema, JsonRowSerializationSchema
from pyflink.datastream.connectors import FlinkKafkaProducer, FlinkKafkaConsumer

class SellTotal(CoFlatMapFunction):

  def open(self, runtime_context: RuntimeContext):
      state_desc = MapStateDescriptor('map', Types.LONG(), Types.STRING())
      self.state = runtime_context.get_map_state(state_desc)
        
      cnt_desc = ValueStateDescriptor('cnt', Types.LONG())
      self.cnt_state = runtime_context.get_state(cnt_desc)
    
  # (id, name)
  def flat_map1(self, value):
      self.state.put(value[0], value[1])
      #return Row(value[0], f"update {value[1]}", 0)

  # (id, cnt)
  def flat_map2(self, value):
      cnt = self.cnt_state.value() or 0
      total = cnt + value[1]
      self.cnt_state.update(total)
        
      if not self.state.contains(value[0]):
          name = "NONAME"
      else:
          name = self.state.get(value[0])

      #return Row(value[0], name, total)
      yield Row(value[0], name, total)


def sell_total_demo(env):
  type_info1 = Types.ROW([Types.LONG(), Types.STRING()])
  ds1 = env.from_collection(
      [(1, 'apple'), (2, 'banana'), (3, 'cherry'), (4, 'durian'), (6, 'fig'), (7, 'grape')],
      type_info=type_info1)

  type_info2 = Types.ROW([Types.LONG(), Types.LONG()])
  ds2 = env.from_collection(
      [(1, 1), (2, 3), (3, 5), (1, 5), (5, 100), (6, 66), (1, 10)],
      type_info=type_info2)

  output_type_info = Types.ROW([Types.LONG(), Types.STRING(), Types.LONG()])
  serialization_schema = JsonRowSerializationSchema.Builder() \
      .with_type_info(output_type_info) \
      .build()
  kafka_producer = FlinkKafkaProducer(
      topic='TempResults',
      serialization_schema=serialization_schema,
      producer_config={'bootstrap.servers': 'kafka:9092', 'group.id': 'test_group'}
  )

  connected_ds = ds1.connect(ds2)
  connected_ds.key_by(lambda a: a[0], lambda a: a[0]).flat_map(SellTotal(), output_type_info).add_sink(kafka_producer)

  env.execute()


if __name__ == '__main__':
    logging.basicConfig(stream=sys.stdout, level=logging.INFO, format="%(message)s")

    env = StreamExecutionEnvironment.get_execution_environment()
    env.add_jars("file:////home/ec2-user/flink-1.15.2/opt/flink-sql-connector-kafka-1.15.2.jar")

    print("connected demo gogo")
    sell_total_demo(env)
```

In `flat_map1` the stream 1 is handled, that is to say, the mapping of item number and item name is maintained, so this stream does not need to generate output events.

The core of the whole application is in `flat_map2`. We take the accumulated quantity from `self.cnt_state` and not only add the new quantity but also update it back to the state. Then, in the output process, we take the corresponding name from `self.state` and finally we output the enriched events.

## Conclusion

In the last example, we demonstrate the operation, state, and merging of streams.

From this example, we can easily understand that Flink can do anything we want as long as we write the program correctly.

We will continue to do some experiments on stream processing and will continue to publish any further updates if there are any.