Stream processing

The HERE platform supports Apache Flink framework for running streaming pipelines. We offer two different modules to do that, Flink Connector and Flink Support.

Note

The Flink Connector currently supports stream, index and volatile layers for read and write and versioned layers for read only.

HERE strongly suggests to use Flink Connector whenever possible as this allows to make use of the full power of Apache Flink framework.

Flink Connector and Flink Support modules can be combined.

Flink connector

Flink Connector implements the standard Flink interfaces that allow you to create source Tables for reading and sink Tables for writing to stream layers.

As a result, you can use both the relational APIs that Flink supports, which are Table API and SQL. Moreover, you can convert a Table to a DataStream and use the Flink DataStream API.

This means, your business logic does not need to contain any HERE-specific function calls.

For detailed explanation of Flink Connector, see Flink Connector.

Flink support

Flink Support is a HERE-proprietary implementation of using data from HERE platform catalogs and layers with Apache Flink framework. The distribution of processing jobs to workers is done by Flink but the data model is a HERE-proprietary format. There is no SQL-like interface, so you cannot select and filter data using Table API.

In the end, FlinkDataClient is a wrapper for regular DataClient making sure that all asynchronous communication is synchronized to make it usable with Flink Source and Sink interfaces. There are no additional functions for tabular access.

For detailed explanation of Flink Support, see Flink Support.

Limitations

In case of working with protobuf messages SdiiMessage.Message, serializer must be registered in the environment:

val serializerConfig = env.getConfig.getSerializerConfig.asInstanceOf[SerializerConfigImpl]
serializerConfig.addDefaultKryoSerializer(classOf[SdiiMessage.Message],
                                          classOf[ProtobufKryoSerializer[SdiiMessage.Message]])

((SerializerConfigImpl) env.getConfig().getSerializerConfig())
    .addDefaultKryoSerializer(
        SdiiMessage.Message.class,
        (Class<? extends com.esotericsoftware.kryo.Serializer<?>>)
            (Class<?>) ProtobufKryoSerializer.class);