DEV Community: Hans-Peter Grahsl

Flink CDC from the Trenches: Handling JSON in Pipelines with UDFs

Hans-Peter Grahsl — Tue, 16 Dec 2025 13:04:41 +0000

Introduction

Recently, I shared a custom function (FROM_JSON) for Apache Flink to ease the handling of complex JSON structures in situations where a deeper integration into Flink's type system is not only beneficial but required. While it's a helpful addition for data engineers, who are primarily building streaming workloads directly with Flink SQL or on top of Flink's Table API, it didn't take long before I received a few questions around the applicability of this very UDF as part of end-to-end data flows that are built with Flink CDC pipeline connectors.

In this article, I share selected learnings and condensed findings I discovered while exploring the most relevant parts of Flink CDC’s codebase to figure out what it takes to implement a FROM_JSON UDF equivalent for Flink CDC pipelines.

My journey involved navigating some interesting aspects and challenges related to differences in the involved type systems, the lack of specific capabilities for user-defined functions in Flink CDC, and some subtleties during the handling of complex, nested data structures, as well as specific binary runtime representations for objects of certain data types.

Motivation

The primary reason why I decided to spend some non-negligible amount of time with Flink CDC lately was triggered by the fact that JSON data is currently not treated as a first-class citizen in Flink CDC pipelines. The following concrete example illustrates what I mean by that. Let's assume we have a Postgres table that contains a few columns, three of which store JSON/JSONB:

postgres=# \d+ public.gov_fake_citizens
                                         Table "public.gov_fake_citizens"
     Column      |  Type   | Collation | Nullable | Default | Storage  | Compression | Stats target | Description
-----------------+---------+-----------+----------+---------+----------+-------------+--------------+-------------
 id              | uuid    |           | not null |         | plain    |             |              |
 personal        | json    |           | not null |         | extended |             |              |
 isactive        | boolean |           | not null | false   | plain    |             |              |
 registered      | text    |           | not null |         | extended |             |              |
 contact         | json    |           | not null |         | extended |             |              |
 knownresidences | json    |           | not null |         | extended |             |              |
Indexes:
    "gov_fake_citizens_pkey" PRIMARY KEY, btree (id)
Access method: heap

Here is how one row in this table might look:

                  id                  |                                                     personal                                                     | isactive |          registered           |                             contact                             |                                        knownresidences                                         
--------------------------------------+------------------------------------------------------------------------------------------------------------------+----------+-------------------------------+-----------------------------------------------------------------+------------------------------------------------------------------------------------------------
 11e66db7-6834-1efe-498c-b7c9d661fb5b | {"firstname":"Emile","lastname":"Sanford","age":28,"eyecolor":"gray","gender":"female","height":163,"weight":51} | f        | 2025-10-17T07:30:49.067+00:00 | {"email":"morgan.shanahan@gmail.com","phone":"+1 305-699-4846"} | ["4361 Prohaska Common, Norikofort, NH 56814","7616 Krajcik Crest, Port Emilioside, ID 36016"]
(1 row)

To run a Flink CDC pipeline that ships data from Postgres to Elasticsearch, the configuration is straightforward. The results, however, are most likely not what the typical friendly data engineer from next door would expect, let alone like to achieve. If we query the Elasticsearch index, the Postgres table row shown above comes back in this way:

{
    "id": "11e66db7-6834-1efe-498c-b7c9d661fb5b",
    "personal": "{\"firstname\":\"Emile\",\"lastname\":\"Sanford\",\"age\":28,\"eyecolor\":\"gray\",\"gender\":\"female\",\"height\":163,\"weight\":51}",
    "contact": "{\"email\":\"morgan.shanahan@gmail.com\",\"phone\":\"+1 305-699-4846\"}",
    "knownresidences": "[\"4361 Prohaska Common, Norikofort, NH 56814\",\"7616 Krajcik Crest, Port Emilioside, ID 36016\"]",
    "isactive": false,
    "registered": "2025-10-17T07:30:49.067+00:00"
}

The reason behind this is that any form of JSON data gets stringified by pipeline source connectors like the Postgres or MySQL ones. From the moment this happens, pipelines just propagate flat strings all the way to the sink. If this is the only result you'd ever need from your Flink CDC pipelines, consider yourself lucky, stop reading, and keep leading an undistracted and happy life.

Chances are, you'd want this to look different and if so, read on to learn more about what needs to happen to end up with documents in your search index that look more like this:

{
    "id": "11e66db7-6834-1efe-498c-b7c9d661fb5b",
    "personal": {
        "firstname": "Emile",
        "gender": "female",
        "weight": 51,
        "eyecolor": "gray",
        "age": 28,
        "lastname": "Sanford",
        "height": 163
 },
    "contact": {
        "phone": "+1 305-699-4846",
        "email": "morgan.shanahan@gmail.com"
 },
    "knownresidences": [
        "4361 Prohaska Common, Norikofort, NH 56814",
        "7616 Krajcik Crest, Port Emilioside, ID 36016"
 ],
    "isactive": false,
    "registered": "2025-10-17T07:30:49.067+00:00"
}

Same, Same But Different

A fundamental aspect to understand when moving from Flink SQL/Table API to Flink CDC context is that while their underlying type systems appear similar on the surface, they are still different from one another. As a consequence, several nuances start to matter, especially when dealing with complex structures and nested data types at runtime, but more on that later.

Schema Definitions

The good news first. Pure schema definitions expressed as string literals are basically interchangeable because both type systems heavily draw from SQL types. We can express the usual primitive suspects and complex types similarly. For instance, the following type definitions are valid in both type systems:

id INT
price DECIMAL(10,2)
hobbies ARRAY<VARCHAR(64)>
scores MAP<STRING,INT>
user ROW<name STRING, age INT, is_active BOOLEAN>

In other words, the high similarity between the two type systems is primarily conceptual and structural. On an implementation level, however, both have their separate Java type equivalents.

Flink Table API defines its types based on classes from:

org.apache.flink.table.types.*
org.apache.flink.table.data.*

Flink CDC defines its types based on classes from:

org.apache.flink.cdc.common.types.*
org.apache.flink.cdc.common.data.*

Schema Parsing

If we were to parse any such schema string literals shown in the above examples into proper data types, the parsing logic would be largely the same. Yet the different Java type systems behind both worlds suggest separate implementations to either express an org.apache.flink.table.types.DataType (Flink Table API) or an org.apache.flink.cdc.common.types.DataType (Flink CDC). Of course, another option would be to parse schema definitions into one of the two type systems only, and complement this with a uni- or bidirectional type converter depending on our needs.

Task 1

Working towards my Flink CDC variant of the FROM_JSON UDF, the initial task was to parse the schema string literal, this time, however, into a DataType for Flink CDC rather than the Flink Table API. By nature, the result is almost identical to the minimum viable schema parser I already had, except it had to respect the different Java types backing Flink CDC’s type system. The first line in the following test method, taken from one of the unit tests, hints at how this piece of code is going to be used:

@Test
@DisplayName("Parse ROW schema with nested ARRAY")
void testRowTypeSchemaWithNestedArray() {

    DataType dataType = SchemaParserCdc.parseType("ROW<id INT, name STRING, hobbies ARRAY<STRING>>");

    // Verify it's a RowType
    assertTrue(dataType instanceof RowType);
    RowType rowType = (RowType) dataType;

    // Verify field count
    assertEquals(3, rowType.getFieldCount());

    // Verify field names
    assertEquals("id", rowType.getFieldNames().get(0));
    assertEquals("name", rowType.getFieldNames().get(1));
    assertEquals("hobbies", rowType.getFieldNames().get(2));

    // Verify field types
    assertTrue(rowType.getFields().get(0).getType() instanceof IntType);
    assertTrue(rowType.getFields().get(1).getType() instanceof VarCharType);
    assertTrue(rowType.getFields().get(2).getType() instanceof ArrayType);

    // Verify array element type
    ArrayType hobbiesArray = (ArrayType) rowType.getFields().get(2).getType();
    assertTrue(hobbiesArray.getElementType() instanceof VarCharType);
}

With the schema parsing out of the way, it’s time to shift our focus to the implementation of the user-defined function itself, which will make use of this schema parser to infer the Flink CDC data type according to any custom target schema.

UDFs that aren’t the UDFs You Know

Having written several basic and advanced UDFs for Flink Table API/SQL as well as other data processing systems such as ksqlDB—yeah I know—and Apache Spark in the past, I initially expected Flink CDC UDFs to offer similar capabilities and the same flexibility. Turns out I was in for a little surprise.

Functions in Flink Table API/SQL

extensive library of built-in system functions
user-defined functions across multiple categories (scalar, table, aggregate, table aggregate, and process table function)
async processing for scalar and table functions
context access in the open(...) lifecycle method
automatic, annotation-based, and custom programmatic type inference, including call context accessibility
comprehensive docs

In short, very mature, powerful, and feature-rich, allowing for various capabilities ranging from the most basic to very advanced ones.

Functions in Flink CDC

relatively small library of built-in system functions
focused on a single function category, namely scalar functions context access in the open(...) lifecycle method but with caveats—more on that follows :)
interoperability with Table API/SQL scalar functions, with notable limitations such as no lifecycle methods or no type hint/type inference compatibility
automatic or programmatic type inference, however, without call context accessibility or other similar advanced aspects
full UDF docs essentially fit on a single screen, which, by the way, is a great opportunity to contribute to this rather promising project!

The bottom line here is that, as it stands today, the UDF support in Flink CDC is, well, lean and seems to primarily focus on simpler use cases than what I had in mind here with this FROM_JSON UDF. Any custom scalar functions we’d write could be either applied in projection and/or filter definitions as part of the pipeline’s respective transform block:

transform:
  - source-table: mydb.mytable
    projection: "*, MY_UDF_A(some_col_1)"
    filter: MY_UDF_B(some_col_2) = 123

pipeline:
  user-defined-function:
    - name: MY_UDF_A
      classpath: com.github.hpgrahsl.flink.cdc.udfs.MyDemoUdfA
    - name: MY_UDF_B
      classpath: com.github.hpgrahsl.flink.cdc.udfs.MyDemoUdfB

The limitations mentioned above create an interesting challenge, though. How or where would we specify the schema definition for a FROM_JSON function call, so that the Flink CDC runtime can properly infer the actual return type when applying this function?

For Flink Table API/SQL functions, this is straightforward, as we can state the schema string literal in a second call parameter and derive the necessary type information directly from that.

SELECT FROM_JSON(my_json_col, 'ARRAY<STRING>') AS parsed_json

Unfortunately, it seems there is no such equivalent that would allow type inference based on the UDF's call context in Flink CDC land, or at least I’m not aware of it. My next idea was to specify the required target schema using some function configuration parameters, which can be accessed in the UDF’s open(...) lifecycle method—far too easy, right? Trying to do so, I figured out that while some classes related to Flink CDC UDFs are prepared for configuration parameters, others are not. Therefore, I had to dig deeper to identify the gaps across about a handful of classes in the current implementation. As a result, I opened FLINK-38792 based on these findings.

Task 2

Since I needed to make progress, I decided to move forward by patching my own fork of Flink CDC 3.6-SNAPSHOT to get proper UDF configuration parameter support. With this in place, I can now do the following:

pipeline:
  user-defined-function:
    - name: MY_DEMO_UDF
      classpath: com.github.hpgrahsl.flink.cdc.udfs.MyDemoUdf
      some.config.param: foo
      another.setting: 123

and can define an arbitrary configuration for my custom functions as part of the pipeline definition. Call me biased, but I definitely feel this is an improvement and probably also the originally targeted behaviour. Thus, I hope that the upstream project will eventually decide to pick up a PR I’m planning to contribute.

Coming back to our original context, I can define the FROM_JSON UDF like so:

pipeline:
  user-defined-function:
    - name: FROM_JSON
      classpath: com.github.hpgrahsl.flink.cdc.udfs.json.FromJsonCdcUdf
      from.json.schema: ARRAY<STRING>

In this particular example configuration, the UDF is instructed to parse JSON arrays containing string elements, which themselves originate from a database column captured by a Flink CDC pipeline connector, for instance, the one for Postgres. The configuration parameter from.json.schema is read in the function’s open(...) lifecycle method and stored in an attribute.

@Override
public void open(UserDefinedFunctionContext context) throws Exception {
    if (context != null && context.configuration() != null) {
        this.configuredSchema = context.configuration().get(SCHEMA_CONFIG);
    }

    if (configuredSchema == null || configuredSchema.trim().isEmpty()) {
        throw new IllegalArgumentException(
            "Schema configuration is required. Set '"
                + SCHEMA_CONFIG.key()
                + "' in the UDF configuration of your pipeline YAML.");
    }
    //...
}

This way, the schema information becomes accessible in the getReturnType() method, which uses the schema parser I previously wrote to create the corresponding DataType. The Flink CDC runtime can thus infer the proper DataType to expect whenever it needs to evaluate the UDF in question.

@Override
public DataType getReturnType() {
    if (configuredSchema != null && !configuredSchema.trim().isEmpty()) {
        return SchemaParserCdc.parseType(configuredSchema);
    }
    throw new IllegalArgumentException(
            "Schema configuration is required. Set '"
                + SCHEMA_CONFIG.key()
                + "' in the UDF configuration (e.g., in pipeline YAML under 'config').");
}

Everything up to this point can be considered as foundational preparations before we can eventually come to the meat of it—converting the actual JSON data.

From JSON to Typed Values for Flink CDC

Walking the JSON structure and converting the input data into properly typed values matching the Flink CDC type system along the way doesn’t sound too complicated, or does it? Well, based on my initial experiments, it turned out not to be a walk in the park and here is why.

RecordData

RecordData is a vital interface as it describes how table rows in Flink CDC are represented by a complex type, the RowType. It also defines how each SQL data type is meant to be mapped to internal data structures. These internal data structures are themselves based on either standard Java types (for primitives) or Flink CDC-specific Java types defined by additional custom interfaces or classes such as StringData, DecimalData and others. The table below is taken directly from the Java docs and contains the full picture:

 * +--------------------------------+-----------------------------------------+
 * | SQL Data Types                 | Internal Data Structures                |
 * +--------------------------------+-----------------------------------------+
 * | BOOLEAN                        | boolean                                 |
 * +--------------------------------+-----------------------------------------+
 * | CHAR / VARCHAR / STRING        | {@link StringData}                      |
 * +--------------------------------+-----------------------------------------+
 * | BINARY / VARBINARY / BYTES     | byte[]                                  |
 * +--------------------------------+-----------------------------------------+
 * | DECIMAL                        | {@link DecimalData}                     |
 * +--------------------------------+-----------------------------------------+
 * | TINYINT                        | byte                                    |
 * +--------------------------------+-----------------------------------------+
 * | SMALLINT                       | short                                   |
 * +--------------------------------+-----------------------------------------+
 * | INT                            | int                                     |
 * +--------------------------------+-----------------------------------------+
 * | BIGINT                         | long                                    |
 * +--------------------------------+-----------------------------------------+
 * | FLOAT                          | float                                   |
 * +--------------------------------+-----------------------------------------+
 * | DOUBLE                         | double                                  |
 * +--------------------------------+-----------------------------------------+
 * | DATE                           | int (number of days since epoch)        |
 * +--------------------------------+-----------------------------------------+
 * | TIME                           | int (number of milliseconds of the day) |
 * +--------------------------------+-----------------------------------------+
 * | TIMESTAMP                      | {@link TimestampData}                   |
 * +--------------------------------+-----------------------------------------+
 * | TIMESTAMP WITH LOCAL TIME ZONE | {@link LocalZonedTimestampData}         |
 * +--------------------------------+-----------------------------------------+
 * | TIMESTAMP WITH TIME ZONE       | {@link ZonedTimestampData}              |
 * +--------------------------------+-----------------------------------------+
 * | ROW                            | {@link RecordData}                      |
 * +--------------------------------+-----------------------------------------+
 * | ARRAY                          | {@link ArrayData}                       |
 * +--------------------------------+-----------------------------------------+
 * | MAP                            | {@link MapData}                         |
 * +--------------------------------+-----------------------------------------+

The main takeaway for successfully converting JSON with a UDF is as follows: for all data types that directly map to a standard Java type, we only need to read the JSON and can extract the corresponding JsonNode values as is without any additional considerations or explicit conversions on top.

Constructed Data Type Caveats

Based on my early-stage experiments while implementing the FROM_JSON UDF, it’s helpful to take a closer look at the three constructed data types.

ArrayData

The ArrayData type is relevant whenever we hit any JSON array node, the corresponding schema literal for which is ARRAY<element_type>. GenericArrayData is an implementation that allows for the convenient construction of objects based on regular Java arrays.

MapData

Any JSON object for which all field values are of the same type can be converted to the MapData type. The schema literal to express this is MAP<key_type,value_type>. GenericMapData implements this interface and can be used for object construction based on regular Java maps. This implementation internally uses two GenericArrayData objects to represent the map’s keys and values.

For both ArrayData and MapData, it’s important to know that all array elements or map entries must themselves be valid internal data structures. What this means is that any array element or map entry that must be represented by means of a special Java type (see table above) must be explicitly converted before constructing GenericArrayData or GenericMapData objects. This is because their constructors use the data as is without performing type checks let alone do (deep) auto-conversions. Hence, it’s our own responsibility to ensure the constructor only ever receives objects of internal data structure types and potentially convert them upfront.

Task 3

Consequently, the FROM_JSON UDF must respect this at all times, as otherwise, we’d face runtime errors and Flink CDC will almost certainly complain with a ClassCastException whenever this constraint gets violated. To give one concrete example here, if we were to mistakenly build a GenericArrayData object based on a regular Java String[] array rather than a StringData[] array, the following would happen when the Flink CDC runtime tries to access and process its elements:

…
Caused by: java.lang.ClassCastException: class java.lang.String cannot be cast to class org.apache.flink.cdc.common.data.StringData (java.lang.String is in module java.base of loader 'bootstrap'; org.apache.flink.cdc.common.data.StringData is in unnamed module of loader 'app')
        at org.apache.flink.cdc.common.data.GenericArrayData.getString(GenericArrayData.java:221)
        at org.apache.flink.cdc.common.data.ArrayData.lambda$createElementGetter$3e58363e$1(ArrayData.java:228)
…

That said, the code for parsing a JSON array node and creating the GenericArrayData object should look something more like this:

private ArrayData parseArray(JsonNode jsonArray, String schemaString) {
    DataType arrayDataType = getCachedSchema(schemaString);
    if (!(arrayDataType instanceof ArrayType)) {
        throw new IllegalArgumentException(
            "specified schemaString must be of type ARRAY - got: " + schemaString);
    }

    ArrayType arrayType = (ArrayType) arrayDataType;
    DataType elementType = arrayType.getElementType();

    if (jsonArray.size() == 0) {
        return new GenericArrayData(new Object[0]);
    }

    Object[] elements = new Object[jsonArray.size()];
    for (int i = 0; i < jsonArray.size(); i++) {
        elements[i] = convertJsonNodeToInternalType(jsonArray.get(i), elementType);
    }

    return new GenericArrayData(elements);
}

Notice the method call to convertJsonNodeToInternalType(...) inside the loop, which guarantees each JSON array element is converted into the proper Flink CDC internal data structure before the resulting GenericArrayData is created and returned. The conversion of map entries to GenericMapData is done in a similar fashion. Besides, this approach ensures that nested arrays (e.g. ARRAY<ARRAY<...>>), nested maps (e.g. MAP<STRING,MAP<STRING,INT>>), or a mix thereof can be implicitly handled as well by means of recursion.

With map and array support in place, let’s focus on the next aspect, namely, how to handle arbitrary JSON objects. Out of habit, I wrongly assumed there must be a similar type duo in place - something like RowData and GenericRowData - however, I was obviously mixing things up with Flink’s Table API, again. Over here in Flink CDC land, we are instead revisiting RecordData.

RecordData

Whenever the FROM_JSON UDF is about to parse arbitrary JSON objects, be it on top-level or somewhere nested, it’s supposed to build a RecordData object matching the specified target schema that is defined with a corresponding ROW type, for instance, ROW<id INT, name STRING, age INT, is_active BOOLEAN>. The relevant implementation of the RecordData interface is BinaryRecordData, which, for reasons of efficiency and performance, is backed by MemorySegment under the hood.

Task 4

Luckily, and in order for us not having to touch any of the lower-level building blocks in that regard, we can rely on the BinaryRecordDataGenerator, a convenience class in Flink CDC’s runtime type utils package. Similar to ArrayData and MapData, it’s important not to forget that each field in a RecordData object must be an internal data structure. With this in mind, the implementation to generate any such BinaryRecordData objects might look as follows:

private RecordData parseObject(JsonNode jsonObject, String schemaString) {
    DataType rowDataType = getCachedSchema(schemaString);
    RowType rowType = (RowType) rowDataType;
    List<String> nestedFieldNames = rowType.getFieldNames();
    int arity = nestedFieldNames.size();

    Object[] nestedFields = new Object[arity];
    for (int i = 0; i < arity; i++) {
        String fieldName = nestedFieldNames.get(i);
        JsonNode fieldValue = jsonObject.get(fieldName);
        DataType fieldType = rowType.getFields().get(i).getType();
        nestedFields[i] = convertJsonNodeToInternalType(fieldValue, fieldType);
    }
    return new BinaryRecordDataGenerator(rowType).generate(nestedFields);
}

This essentially concludes the most relevant implementation aspects of an equivalent FROM_JSON UDF that can work directly within Flink CDC. What’s left are a few more minor tweaks.

Flink CDC’s built-in `DataTypeConverter`

A critically important class that Flink CDC ships to perform type conversions internally is the DataTypeConverter, which you can find the source code for right here.

Across several methods, we can find multiple larger switch blocks to handle the different data types accordingly. In a few of these places it becomes apparent when reading the code that constructed types, especially the ROW type, aren’t receiving the necessary attention they’d need to enable use cases like the FROM_JSON UDF. Here are two of the more pressing ones:

ROW isn’t supported here:

public static DataType convertCalciteRelDataTypeToDataType(RelDataType relDataType) {
    switch (relDataType.getSqlTypeName()) {
    //…
       case ROW:
        default:
        throw new UnsupportedOperationException(
            "Unsupported type: " + relDataType.getSqlTypeName());
    }
}

ROW type is passed through as is here, which may or may not work depending on how the user code created the Object value in the first place:

public static Object convert(Object value, DataType dataType) {
    if (value == null) {
        return null;
    }
    switch (dataType.getTypeRoot()) {
    // …
       case ROW:
        return value;
        // …
    }
}

Besides, it would be beneficial if the methods convertToArray(...) and convertToMap(...) detected any non-internal data structures used for array elements or map entries and automatically converted them whenever reasonably possible. This could ultimately free user code from explicitly fulfilling any such framework internal type constraints.

Task 5

Since I already patched my Flink CDC fork during the work on Task 2 (see further above), I did the same here and fixed these little limitations in the DataTypeConverter. And with that, I reached a point where FROM_JSON could be integration tested for the first time as part of a projection specification in the Flink CDC pipeline definition while processing actual JSON data in-flight.

Applying `FROM_JSON` in Flink CDC Pipelines

We’ve come a long way, so let’s see an exemplary Flink CDC pipeline definition where we are now able to successfully apply FROM_JSON in a projection as part of the transform block. Here, the knownresidences field refers to a JSON column in the Postgres source database table, which contains JSON arrays with string elements.

source:
   # ...

transform:
  - source-table: public.fake_citizens
    projection: id,registered,isactive,FROM_JSON(knownresidences) as knownresidences

sink:
  # ...

pipeline:
  name: my-pipeline
  parallelism: 1
  user-defined-function:
    - name: FROM_JSON
      classpath: com.github.hpgrahsl.flink.cdc.udfs.json.FromJsonCdcUdf
      from.json.schema: ARRAY<STRING>

Bonus Surprise

The curious and informed reader will almost certainly wonder how to apply this function multiple times within the same pipeline. And this is a great question indeed :)

Unfortunately, we must assume every use of FROM_JSON will most likely require its own schema string literal defined in the function configuration. For obvious reasons, this inherently rules out direct re-use of the same function definition. So, how about we introduce multiple "uniquely named references" to FROM_JSON such that each function definition gets its individual schema string literal configured? The example below tries to do that with two named references:

pipeline:
  name: my-pipeline
  parallelism: 1
  user-defined-function:
    - name: FROM_JSON_1
      classpath: com.github.hpgrahsl.flink.cdc.udfs.json.FromJsonCdcUdf
      from.json.schema: ARRAY<STRING>
    - name: FROM_JSON_2
      classpath: com.github.hpgrahsl.flink.cdc.udfs.json.FromJsonCdcUdf
      from.json.schema: MAP<VARCHAR,INT>

Well, it turns out the wish was father to the thought as the Flink CDC runtime currently stumbles over this. More specifically, this clashes with the way the Janino compiler tries to process it under the hood and causes the following exception:

// …
Caused by: org.codehaus.commons.compiler.CompileException: Line 1, Column 76: Redefinition of parameter "__instanceOfFromJsonCdcUdf"
        at org.codehaus.janino.UnitCompiler.compileError(UnitCompiler.java:13080)
        at org.codehaus.janino.UnitCompiler.buildLocalVariableMap(UnitCompiler.java:3791)
// …
        at org.apache.flink.cdc.runtime.operators.transform.TransformExpressionCompiler.lambda$compileExpression$0(TransformExpressionCompiler.java:76)
        ... 37 more

Pragmatic Workaround

Rather than setting out to directly address the root cause, I decided for a pragmatic, yet more than good enough workaround for my needs here. By pre-defining a certain number of static inner classes (e.g. N=16) inside the original UDF class, the function can be applied several (up to N) times in the same Flink CDC pipeline context, thereby elegantly circumventing the redefinition problem we faced above. Finally, a valid pipeline definition when doing so could look like this:

source:
   # ...

transform:
  - source-table: public.fake_citizens
    projection: id,registered,isactive,FROM_JSON_1(knownresidences) as knownresidences,FROM_JSON_2(scores) as scores

sink:
  # ...

pipeline:
  name: my-pipeline
  parallelism: 1
  user-defined-function:
    - name: FROM_JSON_1
      classpath: com.github.hpgrahsl.flink.cdc.udfs.json.FromJsonCdcUdf$_C1
      from.json.schema: ARRAY<STRING>
    - name: FROM_JSON_2
      classpath: com.github.hpgrahsl.flink.cdc.udfs.json.FromJsonCdcUdf$_C2
      from.json.schema: MAP<VARCHAR,INT>

Summary

Building an equivalent FROM_JSON UDF for Flink CDC pipelines revealed certain implementation challenges stemming from differences between Flink CDC and Flink Table API ecosystems. While both systems use similar SQL-based schema definitions, they rely on entirely separate Java type hierarchies, creating subtle but critical runtime considerations. Another aspect worth highlighting is that Flink CDC's UDF capabilities are currently less mature than the ones found in Flink's Table API/SQL, lacking features like call-context-based type inference. This necessitated an alternate approach using configuration parameters to specify target schemas—functionality that required patching Flink CDC itself (documented in FLINK-38792). Probably the most critical implementation detail involves respecting Flink CDC's internal data structure requirements: constructed types like GenericArrayData and GenericMapData don't perform automatic type conversions, so developers must explicitly convert elements to internal types (StringData, DecimalData, etc.) before target object construction. Failing to do so triggers ClassCastExceptions at runtime. Additional challenges included gaps in the DataTypeConverter class for ROW type handling and issues when trying to reuse UDF definitions in Flink CDC's pipeline YAML. The solution for the latter employs static inner classes to enable multiple FROM_JSON invocations with different schemas in a single pipeline, providing a practical path forward despite current framework limitations.

Takeaways

Type system divergence: Flink Table API and Flink CDC have conceptually similar but implementation-wise distinct type systems with separate Java class hierarchies
UDF limitations: Flink CDC's UDF support is lean compared to Table API, lacking advanced features like call context accessibility for type inference
Configuration gap: UDF configuration parameters aren't fully supported in Flink CDC 3.6-SNAPSHOT, requiring custom patches
Internal data structures matter: ArrayData, MapData, and RecordData all require elements to already be converted to internal types (StringData, DecimalData, etc.)—no automatic conversion occurs leading to a ClassCastException pitfall as using standard Java types causes runtime failures when Flink CDC processes the data
ROW type handling: DataTypeConverter has gaps in ROW type support that need addressing to get complex nested structures working properly
Multiple UDF instances of the same type: Static inner classes seem to provide a pragmatic workaround for applying the same UDF multiple times with different configurations in one pipeline

I learned many things while reading through Flink CDC's codebase, implementing this custom UDF, debugging existing issues, fixing my own bugs, adding unit/integration tests, and writing this article. Hope it's useful for others out there who are trying to extend the processing capabilities of Flink CDC by means of UDFs. At the very least, reading this article should help you save a few precious hours in case you'd hit similar issues.

Happy Flink CDCing!

A slightly closer look at MongoDB 5.0 time series collections - Part 1

Hans-Peter Grahsl — Sun, 18 Jul 2021 13:18:11 +0000

Setting the Scene

Recently, at MongoDB.live 2021 one of the bigger feature announcements was the fact that MongoDB version 5.0 introduces so-called time-series collections. The information about it was primarily high-level and also the current documentation doesn't give away some of the details. This is why I decided to dig just a little bit deeper to improve my personal understanding about what is going on behind the scenes, when storing time-series data with this new collection type in MongoDB.

The Past

For several years already, people have been using MongoDB to store their time-series data. Some of them struggled initially and had to learn the hard way, that one doesn’t simply store time-series data as is. The biggest mistake I’ve seen over and over again in the wild was that data hasn’t been stored in an optimized way. What I mean by that is people didn’t invest any further thoughts into proper schema design for their documents, but instead just inserted e.g. raw sensor measurements directly into collections. In almost all cases, doing so eventually led to a lot of storage and processing overhead, unnecessarily large index structures and oftentimes poor performance overall. The way to properly tackle time-series data storage with MongoDB in the past was to apply a schema design trick called the bucket pattern. The main idea behind this pattern is to store several measurements which logically belong together - e.g. data from one specific sensor over a certain period of time - into a single document which contains a bucket holding multiple of these measurements. Since it’s impractical to indefinitely grow one document and its bucket, the application layer sees to it that it starts a new document based on certain thresholds and rules, which are depending on the granularity of time and the ingestion frequency / interval of sensor data. To give a concrete example, there could be a single document and its bucket, which stores all measurements happening every second for one specific hour of the day. This single document would then contain up to 3600 measurements ingested at a 1 second interval during a particular hour of the day, before a new document would then be created to store all the measurements of the same sensor for the next hour of the day.

While this approach can work pretty fine, one needs to invest upfront thoughts regarding schema design and in addition, it means a higher burden for developers. They have to implement as well as tweak and tune the bucketing logic for such time-series ingestion scenarios in the application layer. Also when it comes to certain types of queries there is more effort involved when targeting collections that contain documents which are structured according to the bucket pattern. This is because for queries against such collections, the particular bucketing strategy has to be known and considered accordingly.

The Present

Fast forward to the MongoDB release 5.0 which now brings “native” support for time-series collections. The promise is, that developers don’t need to agonize over schema design tricks such as the bucket pattern any longer. Instead, they can simply insert and query their time-series data directly, without any further considerations on the application layer. But how does this exactly work and how does it look behind the scenes from a document storage perspective?

The following explorations are based on raw measurements. The data contains 3 fields and looks as follows:

{ ts: 2021-07-10T00:00:03.000Z,
  metadata: { sensorId: 31096, type: 'windspeed' },
  value: 32.53987084180961 }

ts represents the timestamp of sensor data
metadata stores which sensor and type of data we are dealing with
value holds the actual sensor reading, a windspeed value in this case

Note that in general, you can have much more complex measurement documents containing more payload fields with varying data types and nested elements, too. It's kept simple here on purpose.

Step 1: Creating a time series collection

The command to create this new time series collection type is as follows:

db.createCollection("windsensors", { timeseries: { timeField: "ts", metaField: "metadata", granularity: "seconds" } } )

Besides the name, we specify time series related settings. Most importantly and the only obligatory config is to provide the name of the field which holds the timestamp of measurements, "ts" in this case. The "metaField" is a descriptive label for the sensor data and the "granularity" (hours, minutes or seconds = default) defines the expected ingestion interval for the sensor readings in question.

Step 2: Insert sample documents

With our empty time series collection in place, let’s ingest the following 10 sample documents, originating from 4 different sensors:

db.windsensors.insertMany([
   {"metadata":{"sensorId":52396,"type":"windspeed"},"ts":ISODate("2021-07-10T00:00:02Z"),"value":18.263742590570686},
   {"metadata":{"sensorId":31096,"type":"windspeed"},"ts":ISODate("2021-07-10T00:00:03Z"),"value":32.53987084180961},
   {"metadata":{"sensorId":52396,"type":"windspeed"},"ts":ISODate("2021-07-10T00:00:03Z"),"value":18.106480571706808},
   {"metadata":{"sensorId":62088,"type":"windspeed"},"ts":ISODate("2021-07-10T00:00:04Z"),"value":20.306831899199864},
   {"metadata":{"sensorId":31096,"type":"windspeed"},"ts":ISODate("2021-07-10T00:00:04Z"),"value":0.6909954039798452},
   {"metadata":{"sensorId":62088,"type":"windspeed"},"ts":ISODate("2021-07-10T00:00:06Z"),"value":0.031065898581725086},
   {"metadata":{"sensorId":27470,"type":"windspeed"},"ts":ISODate("2021-07-10T00:00:07Z"),"value":6.878726412679837},
   {"metadata":{"sensorId":31096,"type":"windspeed"},"ts":ISODate("2021-07-10T00:00:07Z"),"value":3.9089926192773534},
   {"metadata":{"sensorId":52396,"type":"windspeed"},"ts":ISODate("2021-07-10T00:00:07Z"),"value":28.03679268099916},
   {"metadata":{"sensorId":52396,"type":"windspeed"},"ts":ISODate("2021-07-10T00:00:07Z"),"value":1.0575968433736358}
])

Step 3: Run simple find query against time series collection

db.windsensors.find()

The result set shows that all 10 documents are returned separately, which might be surprising at first sight, because this pretty much resembles what we would expect from a "normal" collection, i.e. without any kind of time series optimized storage.

{ ts: 2021-07-10T00:00:02.000Z,
  metadata: { sensorId: 52396, type: 'windspeed' },
  _id: ObjectId("60f3350afbb696c9ace09a19"),
  value: 18.263742590570686 }
{ ts: 2021-07-10T00:00:03.000Z,
  metadata: { sensorId: 52396, type: 'windspeed' },
  _id: ObjectId("60f3350afbb696c9ace09a1b"),
  value: 18.106480571706808 }
{ ts: 2021-07-10T00:00:07.000Z,
  metadata: { sensorId: 52396, type: 'windspeed' },
  _id: ObjectId("60f3350afbb696c9ace09a21"),
  value: 28.03679268099916 }
{ ts: 2021-07-10T00:00:07.000Z,
  metadata: { sensorId: 52396, type: 'windspeed' },
  _id: ObjectId("60f3350afbb696c9ace09a22"),
  value: 1.0575968433736358 }
{ ts: 2021-07-10T00:00:03.000Z,
  metadata: { sensorId: 31096, type: 'windspeed' },
  _id: ObjectId("60f3350afbb696c9ace09a1a"),
  value: 32.53987084180961 }
{ ts: 2021-07-10T00:00:04.000Z,
  metadata: { sensorId: 31096, type: 'windspeed' },
  _id: ObjectId("60f3350afbb696c9ace09a1d"),
  value: 0.6909954039798452 }
{ ts: 2021-07-10T00:00:07.000Z,
  metadata: { sensorId: 31096, type: 'windspeed' },
  _id: ObjectId("60f3350afbb696c9ace09a20"),
  value: 3.9089926192773534 }
{ ts: 2021-07-10T00:00:04.000Z,
  metadata: { sensorId: 62088, type: 'windspeed' },
  _id: ObjectId("60f3350afbb696c9ace09a1c"),
  value: 20.306831899199864 }
{ ts: 2021-07-10T00:00:06.000Z,
  metadata: { sensorId: 62088, type: 'windspeed' },
  _id: ObjectId("60f3350afbb696c9ace09a1e"),
  value: 0.031065898581725086 }
{ ts: 2021-07-10T00:00:07.000Z,
  metadata: { sensorId: 27470, type: 'windspeed' },
  _id: ObjectId("60f3350afbb696c9ace09a1f"),
  value: 6.878726412679837 }

In fact, when we refer to windsensors in our query, we are working with a logical abstraction which is officially deemed to be a "writable, non-materialized view". We can verify this by inspecting the currently existing views as follows. Running

db.getCollection('system.views').find()

shows

{ _id: 'mytsdemo.windsensors',
  viewOn: 'system.buckets.windsensors',
  pipeline: 
   [ { '$_internalUnpackBucket': 
        { timeField: 'ts',
          metaField: 'metadata',
          bucketMaxSpanSeconds: 3600,
          exclude: [] } } ] }

The view definition informs us, that it is based on a collection called system.buckets.windsensors. Other than for "normal", user created views, the pipeline field for this special view shows a placeholder called $_internalUnpackBucket together with the time series related config settings which were used during the creation of the respective collection. Worth noting is the bucketMaxSpanSeconds field which is 3600 here. It is a value in seconds and depends on the chosen granularity which was set during creation time. For this example it means that a bucket would span at most 3600 seconds, i.e. 1 hour. The take away from this is that the actual storage optimized time series data can be found in separate, "internal" collection specified in the viewOn field of the logical view abstraction.

Step 4: Run simple find query against the native underlying collection

Even if there usually shouldn’t be a need to directly access the storage optimized version of the time series data, let’s do it anyway to learn what happens behind the scenes. The following query retrieves just one document from this underlying collection:

db.getCollection('system.buckets.windsensors').findOne()

We get back a result set like this:

{ _id: ObjectId("60e8e30043c83ccb1994f6d5"),
  control: 
   { version: 1,
     min: 
      { ts: 2021-07-10T00:00:00.000Z,
        value: 1.0575968433736358,
        _id: ObjectId("60f3350afbb696c9ace09a19") },
     max: 
      { ts: 2021-07-10T00:00:07.000Z,
        value: 28.03679268099916,
        _id: ObjectId("60f3350afbb696c9ace09a22") } },
  meta: { sensorId: 52396, type: 'windspeed' },
  data: 
   { _id: 
      { '0': ObjectId("60f3350afbb696c9ace09a19"),
        '1': ObjectId("60f3350afbb696c9ace09a1b"),
        '2': ObjectId("60f3350afbb696c9ace09a21"),
        '3': ObjectId("60f3350afbb696c9ace09a22") },
     value: 
      { '0': 18.263742590570686,
        '1': 18.106480571706808,
        '2': 28.03679268099916,
        '3': 1.0575968433736358 },
     ts: 
      { '0': 2021-07-10T00:00:02.000Z,
        '1': 2021-07-10T00:00:03.000Z,
        '2': 2021-07-10T00:00:07.000Z,
        '3': 2021-07-10T00:00:07.000Z } } }

Let’s inspect the document structure by taking a closer look at a subset of the contained fields:

control.min holds the bucket’s lower bound timestamp value which depends on the chosen granularity, additionally the lowest value measured in this bucket and the ObjectId referring to the first entry stored in this document’s bucket.
control.max holds the most recent timestamp value stored in this bucket, additionally the highest value measured in this bucket and the ObjectId referring to the last entry stored in this document’s bucket so far.

Obviously the contained data for both, control.min and control.max is updated on-the-fly as new sensor readings are ingested into this document and its bucket. In general, those two sub-documents would store the min and max values for each field contained in the original measurement’s payload. In our case it was only the value field with a single windspeed measurement.

data is a complex object that holds all the information of every sensor data payload that has been ingested so far. There are just 3 fields in this particular example: the document identifier (_id), the timestamp (ts) and the sensor data (value). If there were more fields in the original measurement document besides just value, they would all be stored here in a similar fashion. The measurements themselves are referred to by using a field name given by the bucket index i.e. 0 .. N for every single measurement. In general the data field would hold sub-documents for all payload fields of the original measurement document.

Based on this single document it is possible to reconstruct every original measurement document which was ever ingested into this bucket, simply by combining the meta field with every 3-tuple, e.g. { id.0, ts.0, value.0 } … { _id.N, ts.N, value.N } taken from the _data field. In general, this would be an N-tuple, since the data field would hold sub-documents for all payload fields of the original measurement document. One concrete example for our sample data resulting in the first original measurement document which was stored in this bucket is:

{
_id: ObjectId("60f3350afbb696c9ace09a19"),
 ts: 2021-07-10T00:00:02.000Z,
 value: 18.263742590570686,
 meta: { sensorId: 52396, type: 'windspeed' }
}

If we inspect the other 3 documents in the underlying storage-optimized collection they look all very similar. The only structural difference of the buckets is that currently, each bucket has a different number of entries, which is exactly as it should be because the 10 original documents originated from 4 different sensors each having a varying number of readings being ingested until that point.

{ _id: ObjectId("60e8e30043c83ccb1994f6d6"),
  control: 
   { version: 1,
     min: 
      { ts: 2021-07-10T00:00:00.000Z,
        value: 0.6909954039798452,
        _id: ObjectId("60f3350afbb696c9ace09a1a") },
     max: 
      { ts: 2021-07-10T00:00:07.000Z,
        value: 32.53987084180961,
        _id: ObjectId("60f3350afbb696c9ace09a20") } },
  meta: { sensorId: 31096, type: 'windspeed' },
  data: 
   { _id: 
      { '0': ObjectId("60f3350afbb696c9ace09a1a"),
        '1': ObjectId("60f3350afbb696c9ace09a1d"),
        '2': ObjectId("60f3350afbb696c9ace09a20") },
     value: 
      { '0': 32.53987084180961,
        '1': 0.6909954039798452,
        '2': 3.9089926192773534 },
     ts: 
      { '0': 2021-07-10T00:00:03.000Z,
        '1': 2021-07-10T00:00:04.000Z,
        '2': 2021-07-10T00:00:07.000Z } } }
{ _id: ObjectId("60e8e30043c83ccb1994f6d7"),
  control: 
   { version: 1,
     min: 
      { ts: 2021-07-10T00:00:00.000Z,
        value: 0.031065898581725086,
        _id: ObjectId("60f3350afbb696c9ace09a1c") },
     max: 
      { ts: 2021-07-10T00:00:06.000Z,
        value: 20.306831899199864,
        _id: ObjectId("60f3350afbb696c9ace09a1e") } },
  meta: { sensorId: 62088, type: 'windspeed' },
  data: 
   { _id: 
      { '0': ObjectId("60f3350afbb696c9ace09a1c"),
        '1': ObjectId("60f3350afbb696c9ace09a1e") },
     value: { '0': 20.306831899199864, '1': 0.031065898581725086 },
     ts: { '0': 2021-07-10T00:00:04.000Z, '1': 2021-07-10T00:00:06.000Z } } }
{ _id: ObjectId("60e8e30043c83ccb1994f6d8"),
  control: 
   { version: 1,
     min: 
      { ts: 2021-07-10T00:00:00.000Z,
        value: 6.878726412679837,
        _id: ObjectId("60f3350afbb696c9ace09a1f") },
     max: 
      { ts: 2021-07-10T00:00:07.000Z,
        value: 6.878726412679837,
        _id: ObjectId("60f3350afbb696c9ace09a1f") } },
  meta: { sensorId: 27470, type: 'windspeed' },
  data: 
   { _id: { '0': ObjectId("60f3350afbb696c9ace09a1f") },
     value: { '0': 6.878726412679837 },
     ts: { '0': 2021-07-10T00:00:07.000Z } } }

Step 5: Playing with bucket growth and bucket limits

The bucket document for meta: { sensorId: 52396, type: 'windspeed' } currently holds 4 sensor readings. The question is how many more measurements can we ingest into this bucket? Obviously, buckets cannot grow indefinitely so there has to be an upper bound.

Earlier, we inspected the view definition of the logical abstraction and briefly mentioned the maxBucketSpanSize setting. When choosing a granularity of seconds during the creation of a time series collection the value for maxBucketSpanSize is 3600. In other words, this means buckets like this can span 1 hour worth of data. If we were to ingest one measurement per second we might assume that such a bucket can store up to 3600 sensor readings, one every second. However, when trying this we see a different behaviour. It seems that there is some kind of fixed upper bound of 1000 entries per bucket in a time series collection. The example document below shows a "full bucket" for sensorId 52396 with their first and last bucket entries respectively while omitting the rest of the data for reasons of brevity.

{ _id: ObjectId("60e8e30043c83ccb19952b3f"),
  control: 
   { version: 1,
     min: 
      { _id: ObjectId("60f2ef27f161e04419383dbb"),
        ts: 2021-07-10T00:00:00.000Z,
        value: 0.13698301036640723 },
     max: 
      { _id: ObjectId("60f2ef3bf161e04419477ec3"),
        ts: 2021-07-10T00:39:23.000Z,
        value: 37.4522108368789 } },
  meta: { sensorId: 52396, type: 'windspeed' },
  data: 
   { value: 
      { '0': 18.106480571706808,

      ...

        '999': 3.6329149494110027 },
     ts: 
      { '0': 2021-07-10T00:00:03.000Z,

      ...

        '999': 2021-07-10T00:39:23.000Z },
     _id: 
      { '0': ObjectId("60f2ef36f161e0441944f2df"),

      ...

      '999': ObjectId("60f2ef2ff161e044193f73ff") 
      } 
   }
}

I haven’t found any indication in the current official documentation about this "magic constant" of limiting buckets to 1000 entries. In this case, it cannot be related e.g. to the document size limit because storing 1000 entries with this sample data doesn’t come anywhere close to a hard document size limit. Maybe the source code would reveal more about this, but so far I didn’t take the time to study the implementation itself.

We can also see from the control.min and control.max timestamps that this particular bucket span size is "only" 2363 seconds which is less than the maximum possible value of 3600. This is because the bucket hit its 1000 entries limit before the span size could be reached. In general, a bucket is closed and a new document created, if either its maxBucketSpanSize is reached or its maximum entries are exceeded (currently 1000), whichever happens first.

Another learning based on these observations explains the recommendation found in the official docs, namely that the chosen granularity settings should match the actual data ingestion rate as closely as possible. In our example of a time series collection with "seconds" granularity the bucket size is 1 hour (3600 sec). If, however, we would only ingest 2 - 3 values per hour this would mean, that we would get many new documents in the underlying time series collection with very small buckets of only 2 - 3 entries each. Clearly, this would drastically impact performance in a negative way and reduce the whole storage optimization mechanism of time series collections to absurdity. So choose the granularity of your time series collections wisely.

Conclusion and Outlook

MongoDB 5.0 introduced a new, natively optimized collection type for storing time series data. It makes the lives of developers easier because working with time series collections is a whole lot easier and more convenient when contrasting this to the past, where it was necessary to explicitly implement the bucket pattern. I hope this article contributed a bit to your understanding about what exactly happens behind the scenes of time series collections from a document storage perspective and the corresponding schema which implicitly reflects the ideas behind the bucket pattern. The most important thing to keep in mind is to take my observations with a grain of salt because it was my first quick exploration of this new MongoDB 5.0 feature.

I plan to write more parts in this series. The 2nd article, should discuss different kinds of aggregation queries over time series collections focusing on the newly introduced window functions.

Stay tuned!

Image Credits:
(c) lukechesser @ Unsplash - https://unsplash.com/photos/JKUTrJ4vK00

My very own perspective on the DevRel scenery

Hans-Peter Grahsl — Wed, 28 Oct 2020 20:46:30 +0000

A while ago, I took the time to kind of sort some of my thoughts about how I have experienced the DevRel scenery in general and the "mission" of developer advocates in particular over the last 3-4 years.

Spoiler: it changed and made me personally struggle

But before pointing out why, let me take one step back and briefly reflect on my very own journey about engaging with developer communities.

Disclaimer: No reader is supposed to take anything of the following personal. Also keep in mind that as a non-native speaker I might not be able to express myself in the best possible way 🙃

Over the last couple of years I tried to become ever more active in two vastly growing developer communities. 2015-2017, it began with various activities within MongoDB territories. A little later, starting with 2018, my focus shifted a bit towards the Apache Kafka ecosystem and its community. It basically stayed like this until today and being an active member in both these communities gave me lots of opportunities to learn from others and thereby personally grow in various areas. While the public image might be a different one, my self-perception combined with anecdotal evidence is, that along the way, I managed to become not only a passionate but recognized community member. Adding to that, I would even go one step further and characterize myself as a "self-taught and independent developer advocate", for both these tech stacks today. Whether this makes some readers laugh or just smile is not really relevant at this point.

Coming back to developer relations, it's clear that a large part of professional advocates' work is of course to engage with the developer community across various different touch points. First and foremost, at on-site or virtual events, which might be differentiated as follows:

1) There are big stage events run by vendors which are largely focused on their very own tech stack and ecosystem

2) Then there are big stage events with a rather broad technology mix, typically independent of specific vendors, apart from sponsorship of course.

Both these types of conferences can be considered to be mostly revenue driven at least from the point of the organizers which is why I would call them "industry events".

3) There are small(er) stage events which are also featuring a nice technology mix.

4) Finally, there are little community events / meetups which are usually the only "free for participants" on-site events and these are at best "cost-covering" but are definitely not profit-oriented.

Your mileage may vary on this "categorization" and I admit that it could have been done in a more nuanced way. Still it's not completely wrong or useless - at least sufficient or good enough to better understand the point I'm trying to make.

The thing is, developer advocates, technology evangelists or however you want to call these folks, originally focused on the two big stage conference types that I mentioned above. In fact, this makes a lot of sense since after all, this is where vendors are likely to generate the most impact business-wise, provided their people are doing a good job during said events. So besides sending out people from their very own DevRel team to engage with the audience at vendor conferences, they are also expected to be seen at big stage multi-tech events. This is totally understandable and perfectly fine.

Over time, however, I've experienced that more and more companies are trying to "flood" - sorry for this unintended negative connotation - smaller, community-driven events with their advocates, too. And herein lies the problem: this is a pain point for me, thus, it gives me a bit of frustration in terms of my very own devrel activities. After all, it's very hard to "prevail over professional advocates" for individuals like myself, especially at smaller to medium-sized community events featuring a mixed technology content. The reasons for this are clear: first, there are typically only one or at best two talks for any given technology at such events and second, given two equally good abstracts, organizers would almost certainly decide for "official advocates" representing the company behind a specific tech rather than for "someone just doing stuff" in the community. It attracts more people and helps them sell tickets which is again understandable from an economic perspective. At the same time, the probabilities to get speaking engagements for folks like me are vastly decreasing, as ever more professional developer advocates are hired and sent out to talk at smaller events, all the way down to meetups.

The thing is, I'm not even trying to change this situation because, as it stands today, I couldn't do much about it anyway. Basically all I can do is "fight and push harder" to still be able to speak here and there, despite the challenges mentioned above. I think you can well imagine that continuing to do so is very exhausting to say the least. The other strategy is to stop doing it and invest the "saved" time & effort for other things. The latter would mean that I'm personally not affected any longer but it would still remain the same, unsatisfactory situation for other community passionates like myself who are "suffering similar pain".

I'm very looking forward to receiving any input on that matter. If you think I'm crazy or an idiot fair enough, but at least please bring something constructive to "my table".

Yours truly.

How to build a streaming 🤩 emojis 😍 tracker app with 🚀 ksqlDB 🚀

Hans-Peter Grahsl — Tue, 31 Mar 2020 18:21:24 +0000

I recently created and published a tiny but handy project which provides a few custom user-defined functions (UDFs) to work with emojis in ksqlDB. While there is a minimum viable amount of project documentation, it's often much easier based on a little application to quickly understand how to make use of these functions and what you can build with them.

The goal is to build a simple, yet fault-tolerant and scalable stream processing backend to do near real-time tracking of emojis based on public tweets. Such an application can be used to feed a live web dashboard which, for instance, might show a continuously updated ranking of the most popular emojis derived from extracting and aggregating emoji occurrences found in tweets. It could look like the illustration below:

Thanks to ksqlDB, we can build all vital components for such an application in one coherent technology stack. Its unified streaming architecture:

allows configuration-based data ingress and egress based on Apache Kafka Connect
enables transformations, aggregations, and enrichments of data with a convenient SQL-like language
and supports both, streaming queries as well as point-in-time lookups against the managed state based on materialized views

Let's get this started step-by-step...

Step 1: Data Ingestion

Directly from within ksqlDB, we can manage Apache Kafka Connect source and sink connectors. For this example, we leverage a turn-key ready Twitter source connector from the community. Before you can deploy this connector you have to get your access credentials for the Twitter API which need to be configured for the source connector. The following ksql snippet - source connector configuration in the WITH clause - does the job and brings in public live tweets that match certain keywords:

CREATE SOURCE CONNECTOR `my-twitter-src-01` WITH (
    'connector.class'='com.github.jcustenborder.kafka.connect.twitter.TwitterSourceConnector',
    'value.converter'='org.apache.kafka.connect.json.JsonConverter',
    'value.converter.schemas.enable'=false,
    'key.converter'='org.apache.kafka.connect.json.JsonConverter',
    'key.converter.schemas.enable'=false,
    'twitter.oauth.accessToken'='...',
    'twitter.oauth.consumerSecret'='...',
    'twitter.oauth.consumerKey'='...',
    'twitter.oauth.accessTokenSecret'='...',
    'kafka.status.topic'='tweets',
    'process.deletes'=false,
    'filter.keywords'='coronavirus,2019nCoV,SARSCoV2,covid19,cov19'
);

NOTE: You have to replace all 4 twitter.oauth.* settings with your access credentials for this to work properly.

Shortly after the source connector starts it will produce JSON records into the configured Apache Kafka topic named tweets. To quickly inspect the data ingestion we define our base stream reduced to the essence of the original payload like so:

-- 'create stream from tweets topic only using a few relevant payload fields'
CREATE STREAM tweets
    (ID BIGINT, CREATEDAT BIGINT, TEXT VARCHAR,LANG VARCHAR,USER STRUCT<SCREENNAME VARCHAR>)
WITH (kafka_topic='tweets',value_format='JSON');

To inspect the output of this stream for a single ingested tweet we SELECT just one record:

ksql> SELECT ID,TEXT FROM tweets EMIT CHANGES LIMIT 1;
+-------------------------------------+---------------------------------------------+
|ID                                   |TEXT                                         |
+-------------------------------------+---------------------------------------------+
|1244971755531354112                  |RT @Faytuks: #BREAKING🚨 - Sweden has confirm|
|                                     |ed 34 new #coronavirus deaths, raising the co|
|                                     |untry's death toll to 180 with 4 028 confirme|
|                                     |d cas…                                       |
Limit Reached
Query terminated

Step 2: Data Processing

Now, for the actual stream processing on top of the ingested data, we first create a derived stream tweets_emojis by applying our custom UDF EXTRACT_EMOJIS. You can find a short description how to install these custom emoji handling functions into your ksql-server installation here.

-- 'create derived stream using the EMOJIS_EXTRACT function on the TEXT column'
CREATE STREAM tweets_emojis AS 
    SELECT EMOJIS_EXTRACT(TEXT,false) AS EMOJIS FROM tweets EMIT CHANGES;

This UDF extracts an array of potentially contained tweets from the raw TEXT column. The 2nd parameter of type boolean allows us to choose between list (unique=false) or set semantic (unique=true) which decides whether or not the extracted array may contain duplicate emoji entries.

The next step is to flatten the emojis array which we can easily achieve by applying ksqlDB's built-in UDTF called EXPLODE. Doing so will automatically skip any tweet which didn't contain emojis at all - those which are represented with an empty array in the above tweets_emojis stream.

-- 'create flattened stream to get each contained emoji of a tweet in a separate record'
CREATE STREAM tweets_emojis_flattened AS 
    SELECT EXPLODE(EMOJIS) AS EMOJI FROM tweets_emojis EMIT CHANGES;

The resulting stream contains all occurring emojis as separate records:

ksql> SELECT ROWKEY,EMOJI FROM tweets_emojis_flattened EMIT CHANGES LIMIT 10;
+--------------------------------+--------------------------------+
|ROWKEY                          |EMOJI                           |
+--------------------------------+--------------------------------+
|{"Id":1244971755531354112}      |🚨                              |
|{"Id":1244971755996880897}      |😢                              |
|{"Id":1244971756122550272}      |💡                              |
|{"Id":1244971757011902469}      |👇🏻                              |
|{"Id":1244971760203845632}      |📽                              |
|{"Id":1244971760203845632}      |☣                              |
|{"Id":1244971760203845632}      |🇷🇺                              |
|{"Id":1244971760203845632}      |🛑                              |
|{"Id":1244971760639827968}      |❤                               |
|{"Id":1244971760287731718}      |🤣                              |
Limit Reached
Query terminated

This stream serves as input to create a table by doing a simple aggregation to group and count the emojis:

-- 'create a table which continuously calculates the total count of each specific emoji'
CREATE TABLE tweets_emoji_counts AS 
   SELECT EMOJI,COUNT(*) AS TOTAL_COUNT FROM tweets_emojis_flattened
      GROUP BY EMOJI EMIT CHANGES;

Step 3: Queries

ksqlDB provides two different types of queries against tables:

Pull Queries

Pull queries can be used to do point-in-time lookups against the continuously updated state which is represented by the table. We can retrieve the current total counter of any specific emoji by specifying it as the ROWKEY in the WHERE clause:

-- 'pull query to retrieve single point-in-time emoji count'
SELECT EMOJI,TOTAL_COUNT FROM tweets_emoji_counts WHERE ROWKEY = '🙏';

ksql> SELECT EMOJI,TOTAL_COUNT FROM
tweets_emji_counts WHERE ROWKEY = '🙏';
+--------------------+--------------------+
|EMOJI               |TOTAL_COUNT         |
+--------------------+--------------------+
|🙏                  |20                  |
Query terminated

Push Queries

Push queries are meant to run indefinitely and emit results every time there are changes in the underlying data stream or updates of the table state respectively. We can subscribe to a continuously updated total counter stream of a single, several specific or all emojis:

-- 'PUSH query: get change stream for single emoji counter'
SELECT EMOJI,TOTAL_COUNT FROM tweets_emoji_counts
   WHERE emoji = '🙏' EMIT CHANGES;

ksql> SELECT EMOJI,TOTAL_COUNT FROM tweets_emoji_counts
WHERE emoji = '🙏' EMIT CHANGES;
+---------------------+---------------------+
|EMOJI                |TOTAL_COUNT          |
+---------------------+---------------------+
|🙏                   |55                   |
|🙏                   |56                   |
|🙏                   |59                   |
|🙏                   |61                   |
^CQuery terminated

-- 'PUSH query: change stream for several emoji counters - there is no IN(...) yet :('
SELECT EMOJI,TOTAL_COUNT FROM tweets_emoji_counts
   WHERE emoji = '🙏' OR emoji = '🚨' OR emoji = '😭' EMIT CHANGES;

ksql> SELECT EMOJI,TOTAL_COUNT FROM tweets_emoji_counts 
WHERE emoji = '🙏' OR emoji = '🚨' OR emoji = '😭' EMIT CHANGES;
+---------------------+---------------------+
|EMOJI                |TOTAL_COUNT          |
+---------------------+---------------------+
|😭                   |48                   |
|🙏                   |66                   |
|🚨                   |109                  |
|😭                   |49                   |
|🚨                   |111                  |
|😭                   |50                   |
|🚨                   |112                  |
|🙏                   |67                   |
^CQuery terminated

Step 4: Serving Data to Clients

Thanks to its HTTP / REST API, ksqlDB lets us directly expose both, push and pull queries to consuming clients. For instance, we can run the needed queries to build a live dashboard in which we visualize the data streams and offer end-users a simple form-based query mechanism. Below is a cURL snippet, which when run against the REST API, is showing the output of a PULL query to retrieve the current counter for the 🙏 emoji:

curl -s -X "POST" "http://localhost:8088/query" \
  -H "Content-Type: application/vnd.ksql.v1+json; charset=utf-8" \
  -d $'{
   "ksql": "SELECT EMOJI,TOTAL_COUNT FROM tweets_emoji_counts WHERE ROWKEY = \'🙏\';",
    "streamsProperties": {}
  }' | jq '.[1] | {emoji: .row.columns[0],count: .row.columns[1]}'

which for my particular data results in:

{
  "emoji": "🙏",
  "count": 2198
}

Have fun handling Emojis in 🚀 ksqlDB 🚀

Towards a MongoDB-backed Apache Kafka Streams State Store

Hans-Peter Grahsl — Mon, 23 Dec 2019 16:03:50 +0000

Setting the Scene

Nowadays, there is an increasing number of business-critical services across different industries that need to process data quicker than ever to be able to react to changing conditions with very low latencies. Irrespective of the concrete technology stack we use for near real-time stream processing, there are two broad categories of operations that we want to apply on never-ending streams of data, both of which are vastly different:

First, some applications process single messages, one by one in a stateless fashion. Examples would be to filter records according to pre-defined criteria or to apply a set of transformations on every message that passes by. Such types of workloads are straight-forward to reason about since their stateless nature lets us process a message independently from any other message that we have already acted upon or might do so in the future.
Second, there are applications which are a completely different "beast" since they need to deal with state management along the way. Examples for stateful operations in the context of stream processing are aggregations, counting, joining streams of data or keeping track of messages within certain timespans a.k.a windowing. The hardest parts to get right in this regard are related to fault-tolerance and scalability of said state.

So yes, efficient and reliable state management is a rather tricky undertaking comprising aspects like state growing very large - beyond the available memory - and at the same time guaranteeing that the maintained state stays consistent and isn't lost in case of certain failure scenarios.
Luckily, most of these challenges are typically taken care of to a very large degree by the stream processing frameworks and technologies that we use - so don't be scared away. Instead, let's take a brief look at how Apache Kafka Streams is dealing with state and how it helps to take off most of the burden from our shoulders.

In case you are already familiar with state management in Kafka Streams applications feel free to skip the following section and dive right into this blog post's matter.

State Management in Apache Kafka Streams

Whenever our stream processing applications leverage stateful operations Kafka Streams takes care to create and continuously maintain the underlying state stores to persist any changes regarding the state. In general, a single application may require one or more such state stores which can be accessed by their respective names to selectively write to and read from the state store in question.
The actual storage mechanism behind the scenes has been built with a 'pluggable architecture' in mind. Out of the box, Kafka Streams supports state stores running on top of RocksDB, which is "an embeddable persistent key-value store for fast storage". Embeddable means, that there is no need to spin-up and run external processes - as we would do for database servers - but instead, RocksDB is used as part of the Kafka Streams application and it typically persists to a locally attached disk of the corresponding processing instance. Besides leveraging RocksDB, we can opt for an in-memory only state store implementation, which essentially means that all state is kept and maintained by a hash map structure held inside the JVM's heap. The application then doesn't persist any state to its locally attached disk. Kafka Streams applications can be scaled-out, such that several processing instances are run in parallel to share the workload. Each of the instances then processes a subset of all available partitions of the involved Kafka topics. Likewise, each of the instances locally holds and maintains a partial application state only. In other words, the entire application state is distributed across multiple nodes. Since nodes can go down anytime, either unexpectedly or on purpose, the instance-local state can, of course, be temporarily lost. To cope with such situations and other failure scenarios, Kafka Streams not only performs state changes locally but writes all modifications into so-called changelog topics backed by Apache Kafka. This allows recovering from state loss because the missing state can be re-created by reprocessing the records from the corresponding Kafka topic partitions.

Handling State Store Access

From a developer's perspective, the way we create state stores and interact with them very much depends on which of the two different Kafka Streams APIs (Streams DSL vs. Processor API) we choose to write our stream processing applications. The DSL offers a very convenient way to define stream processors thanks to its declarative, functional and fluent API nature. Most users like DSL abstractions as well as higher-level building blocks and thus tend to primarily leverage the DSL. More importantly, when it comes to state management due to the use of stateful operators, the DSL takes care to auto-create the needed state stores. Additionally, the interactions with the state stores are handled in a fully transparent manner. That said, all the complexities that come along with explicitly touching state are more or less hidden away.

On the other hand, there are use cases for which we need certain functionality that simply isn't provided by the DSL. Whenever we reach that point, we can "fallback to" and use the imperative-style Processor API (PAPI). While it offers us more flexibility and lower-level building blocks, at the same time, it means we need to write more code ourselves. Also when it comes to state management, it's up to us to create any required state stores in the first place, and explicitly write to or read from them as needed.

Exposing Managed State

More often than not, the application state which is managed by Kafka Streams needs to be made available to the outside world, such that other services can directly perform state-based queries against the state at any point in time. For that purpose, there is a mechanism called Interactive Queries which allows performing simple queries based on a single or a range of keys. Additionally, Kafka Streams maintains and exposes the necessary meta-data which is required to derive which of potentially many processing instances is currently responsible for what parts of the distributed application state. This meta-data can be leveraged to route key-based lookup queries to the right target instance which locally holds and maintains the partial state containing the data for given keys.

The Case for Custom State Stores

While simple key-based state lookups might serve lots of use cases just fine, there are others for which such access patterns could be too restrictive. We might need to support richer queries against the application's state which is potentially given by rather complex aggregate structures that are stored for different keys. Luckily, we learned above that there is a 'pluggable' state store architecture in place and exactly in this regard, let's briefly explore how we could allow for arbitrary queries against the persisted state using a custom, remote state store implementation.

When we think about the implementation of custom state stores, the following high-level characteristics are rather important:

the data store backing the Kafka Streams state store should be resilient & scalable enough and offer acceptable performance because Kafka Streams applications can cause a rather high read/write load
since application state may consist of complex aggregate structures, support for rich querying and indexing capabilities would be beneficial to be able to go way beyond key-based lookups
ideally, the custom state store implementation works as a drop-in replacement so that it can be used with minimal to no code and/or configuration changes for existing Kafka Streams applications

MongoDB-backed State Store Implementation

MongoDB can be considered one of the leading document-oriented databases, which are a very good fit to store, query and index complex aggregate structures. Furthermore, MongoDB offers replication and sharding out-of-the-box which are must-have features to cope with potentially high read/write loads as well as the proper level of resilience for backing Kafka Streams applications' state.

Challenge

Ideally, we would just take any application-level data structures which are used behind arbitrary stateful operations and put them into MongoDB collections as documents. However, that is easier said than done since the Kafka Streams DSL strictly limits the possibilities in that regard. The way it works under the covers is that the DSL uses several "nested stores", wrapping one another, similar to a decorator approach.

The outer-most state store "layer", for instance, is a MeteredKeyValueStore. Then it depends, whether or not caching and/or change logging are enabled. If so, there are e.g. CachingKeyValueStore and ChangeLoggingKeyValueStore in-between, before we finally reach the actual store specific implementation - the RocksDBStore per default or a custom one.

The following stack trace based on debugging an application using the default RocksDBStore also reflects, how this looks like if both, caching and change logging are enabled:

For putting a new key-value pair into the state store:

For getting the value for a key from the state store:

Now, the main challenge about this is the fact, that very early in this series of "decorated" state stores - in fact right after the MeteredKeyValueStore layer - the API contracts are already bound to keys and values being bytes. As a consequence, we do not have direct access anymore to our original data structures (POJOs) at the inner-most "layer", namely the concrete and persisting key-value store implementation. Instead, we have to deal with the bytes resulting from after the serialization which took place in the MeteredKeyValueStore based on the configured Serdes. Leaving aside the DSL, and sticking to state stores that can only be used together with the PAPI, would allow us to implement a fully customizable state store. Doing so, we could easily break free from being tightly bound to handling bits & bytes. But then again, it is not feasible any longer to use the custom state store for any existing code in the wild which relies on the DSL. Gone would then be the option to have a nicely working drop-in replacement.

That being said, the proposed way to deal with this dilemma is to embrace the API constraints and come up with an alternate solution that can work based on the bytes contract. Obviously, the custom state store implementation needs to be aware of the configured Serde classes. It can then leverage the deserializers to restore the original key/value application-level data structures, convert these into MongoDB documents and write them into corresponding collections. Conversely, whatever needs to be read back from the database, is first mapped from documents into the application-level data structures, before being serialized and sent through the state store hierarchy as bytes. The illustration below depicts the handling of data on its way in and out of the database and through the state store hierarchy:

Resulting Sample State Documents in MongoDB

Many more Possibilities

Once we have our stream processing state persisted in MongoDB, we can leverage its full potential when it comes to exposing said state. Any of the available fields can be efficiently queried. Thanks to sophisticated secondary index support this can range from geospatial queries all the way through to full-text search or even complex aggregation pipeline queries if needed. For the simple scenario shown above, we might want to define indexes on the fields avg, min, max and stationName to efficiently support exact matches, range queries and sorting on any of these fields.

Furthermore, MongoDB's change streams feature can be combined with the reactive database driver to directly stream any state changes to 3rd party clients as they happen. A commonly found use case for this would be to feed a live dashboard in a single page application with either all or a specific subset of the state changes that are happening in Kafka Streams applications.

Trade-Offs

De/Serialization Overhead

Without a doubt, the suggested approach to implement a custom state store on top of MongoDB involves de/serialization overhead. On the plus side, it enables a drop-in replacement and can be leveraged together with any existing Kafka Streams DSL code that needs key-value state stores. The same approach could be used to plug-in various other state store backends with relative ease.

Remote State

Having a centralized, remote state store means that our stream processing application has to reach across network boundaries to read/write state, which may or may not be an issue depending on the provided network reliability as well as throughput needs. Furthermore, we shouldn't forget about the availability guarantees the centralized, remote state store can provide, since having the state available at more or less "all times" is vital. Network blips might be neglected and mitigated by retry mechanisms but longer network-induced downtimes of a critical subset of the database nodes would almost certainly cause us trouble.

However, all of these trade-offs depend on the performance characteristics of the underlying infrastructure: in some systems, network delay may be lower than disk access latency, and network bandwidth may be comparable to disk bandwidth. There is no universally ideal trade-off for all situations, and the merits of local versus remote state may also shift as storage and networking technologies evolve. -- Martin Kleppmann

Main Benefits

Despite the main drawbacks described above, there are a number of beneficial aspects that come along with a custom, remote state store implementation which are worth mentioning:

persistence of state can be scaled and replicated independently from the Kafka Streams application nodes
there is no such thing as local state loss if any of the Kafka Streams application nodes goes down
no need to run stand-by replicas to shorten otherwise long-lasting state store rebuilding phases
managed state often reflects rich data structures (aggregates that can be efficiently served to any other application based on arbitrary non-key queries backed by secondary indexes
no additional processing load / network saturation caused by lots of interactive queries which have to be directly served by the Kafka Streams applications nodes themselves
MongoDB can be configured to run on top of the in-memory storage engine to completely avoid disk I/O

Summary & Outlook

This blog post explained the importance of managing state in stream processing applications. While the default RocksDB-backed Apache Kafka Streams state store implementation serves various needs just fine, some use cases could benefit from a centralized, remote state store. In particular, one possible solution for such a customized implementation that uses MongoDB has been discussed. A major drawback of this approach is, of course, the fact that any state-related read and write access has to be done over the network. However, its main benefit is that complex state aggregates can be efficiently served to any 3rd party applications based on rich querying options and indexing support directly from the database, without the need to cause interactive-query-induced CPU/network load on the actual stream processing nodes.

Currently, there is only proof-of-concept code for the MongoDB-backed state store implementation. The idea is to open source it after polishing some rough edges, adding at least basic documentation and further tests. In case you would find that useful or even want to contribute, please just let me know so that I can ping you once the repo is public. Stay tuned!

Many thanks to Matthias J. Sax for providing me very helpful insights to better understand the constraints and trade-offs of remote state stores. Also I'm very thankful that Gunnar Morling took the time to challenge different aspects of this blog post and to provide very useful hints and comments on this work.

DEV Community: Hans-Peter Grahsl

Flink CDC from the Trenches: Handling JSON in Pipelines with UDFs

Introduction

Motivation

Same, Same But Different

Schema Definitions

Schema Parsing

Task 1

UDFs that aren’t the UDFs You Know

Functions in Flink Table API/SQL

Functions in Flink CDC

Task 2

From JSON to Typed Values for Flink CDC

RecordData

Constructed Data Type Caveats

ArrayData

MapData

Task 3

RecordData

Task 4

Flink CDC’s built-in DataTypeConverter

Task 5

Applying FROM_JSON in Flink CDC Pipelines

Bonus Surprise

Pragmatic Workaround

Summary

Takeaways

Happy Flink CDCing!

A slightly closer look at MongoDB 5.0 time series collections - Part 1

Setting the Scene

The Past

The Present

Step 1: Creating a time series collection

Step 2: Insert sample documents

Step 3: Run simple find query against time series collection

Step 4: Run simple find query against the native underlying collection

Step 5: Playing with bucket growth and bucket limits

Conclusion and Outlook

Stay tuned!

My very own perspective on the DevRel scenery

How to build a streaming 🤩 emojis 😍 tracker app with 🚀 ksqlDB 🚀

Step 1: Data Ingestion

Step 2: Data Processing

Step 3: Queries

Pull Queries

Push Queries

Step 4: Serving Data to Clients

Have fun handling Emojis in 🚀 ksqlDB 🚀

Towards a MongoDB-backed Apache Kafka Streams State Store

Setting the Scene

State Management in Apache Kafka Streams

Handling State Store Access

Exposing Managed State

The Case for Custom State Stores

MongoDB-backed State Store Implementation

Challenge

For putting a new key-value pair into the state store:

For getting the value for a key from the state store:

Resulting Sample State Documents in MongoDB

Many more Possibilities

Trade-Offs

De/Serialization Overhead

Remote State

Main Benefits

Summary & Outlook

Flink CDC’s built-in `DataTypeConverter`

Applying `FROM_JSON` in Flink CDC Pipelines