How We Cut PyFlink Pipeline p99 Latency from 3-5 Seconds to ~500ms

The Problem: Our p99 Was 3-5 Seconds

Our PyFlink pipeline was missing its latency SLO by seconds. The pipeline itself was straightforward: consume events from Kafka, transform them, serialize them as Protobuf, and write the results to downstream systems. Yet under production load, p99 end-to-end latency was consistently in the 3-5 second range.

Profiling pointed us to an unexpected bottleneck: we were deserializing Protobuf messages in Python, even though the Flink runtime processing our stream was JVM-based. Every record that entered the Python path had to cross the JVM-to-Python process boundary, get parsed by a Python UDF, and then cross back. The business logic wasn't the problem. The doorway was.

We moved Protobuf deserialization to Flink's JVM-side Protobuf format and kept Python for orchestration and SQL. In our environment, p99 dropped to approximately 500 milliseconds, with less code and a pipeline that is easier to reason about. Verified on AWS Managed Service for Apache Flink (formerly Kinesis Data Analytics).

Why Python-Side Deserialization Is So Expensive

The naive PyFlink architecture looks like this:

1. A Kafka source table declared with a generic format (raw, json, or a SimpleStringSchema), so every record arrives as opaque bytes or a string.
2. A Python map() or UDF that imports generated _pb2.py classes and calls ParseFromString() on every message.
3. Downstream transforms and sinks.

Two costs hide in step 2, and they compound at high throughput.

The process boundary. PyFlink is not Python running inside Flink; it is a JVM runtime coordinating with a separate Python execution environment. Every record that enters the Python execution path incurs overhead associated with moving data between the JVM and Python, and depending on the operator and execution mode, that can involve serialization and inter-process communication in both directions. For a per-record deserialization UDF on a latency-sensitive pipeline, that overhead is paid before the actual business transformation begins.

Per-record parse cost. Even when Python's Protobuf implementation uses its native backend, parsing in a Python UDF still requires the record to enter the Python execution path. When the workload is latency-sensitive and high-throughput, the combination of serialization, inter-process communication, Python execution, and parsing overhead can become significant. In our case, profiling showed that this path was a major contributor to our latency.

In our pipeline, these two costs together accounted for the bulk of the gap between a 3–5 second p99 and the ~500ms target we needed, before the enrichment logic even began executing.

The Key Realization: PyFlink Already Runs on the JVM

Here's the insight that changes the architecture: if Protobuf is declared at the table DDL level, Flink's Kafka connector deserializes it with its native, optimized JVM-based Protobuf format before any data reaches the Python side. Your columns simply arrive typed and ready. Python's role shrinks to what it's genuinely good at in this stack: orchestration and SQL.

No rewrite to Java. No change to how jobs are deployed. Just a different declaration of intent.

The trade is that Flink's native Protobuf format needs compiled Java message classes on the classpath; it does not consume .proto files or Python _pb2 modules directly. That means adding a small build step to your workflow, which we'll cover below.

Implementation

The pipeline splits into two declarative jobs.

Job 1: JSON In, Protobuf Out

The source table reads the raw JSON topic; the sink table declares format = 'protobuf' and points at the compiled Java class. The JVM handles typed-row-to-Protobuf encoding.

-- SOURCE: raw JSON payload as STRING plus Kafka record timestamp
CREATE TABLE source_events_json (
  event_data STRING,
  kafka_timestamp TIMESTAMP(3) METADATA FROM 'timestamp'
) WITH (
  'connector' = 'kafka',
  'topic' = '${INPUT_JSON_TOPIC}',
  'properties.bootstrap.servers' = '${KAFKA_BOOTSTRAP_SERVERS}',
  'scan.startup.mode' = 'latest-offset',
  'format' = 'raw'
);

-- SINK: Protobuf out to Kafka (JVM handles typed row to Protobuf)
CREATE TABLE sink_events_pb (
  id STRING,
  organization_id STRING,
  event_ts ROW<`seconds` BIGINT, `nanos` INT>,
  is_active BOOLEAN,
  event_type STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'acme.events.pb.v1',
  'properties.bootstrap.servers' = 'kafka:9092',
  'format' = 'protobuf',
  'protobuf.message-class-name' = 'com.acme.events.v1.EventOuterClass$EnrichedEvent'
);

-- TRANSFORM: pure SQL, no Python UDFs
INSERT INTO sink_events_pb
SELECT
  JSON_VALUE(event_data, '$.id')              AS id,
  JSON_VALUE(event_data, '$.organization_id') AS organization_id,
  ROW(
    UNIX_TIMESTAMP(NULLIF(JSON_VALUE(event_data, '$.after.event_ts'), '')),
    CAST(EXTRACT(NANOSECOND FROM CAST(NULLIF(JSON_VALUE(event_data, '$.after.event_ts'), '') AS TIMESTAMP_LTZ(9))) AS INT)
  ) AS event_ts,
  CAST(JSON_VALUE(event_data, '$.is_active') AS BOOLEAN) AS is_active,
  JSON_VALUE(event_data, '$.event_type')      AS event_type
FROM source_events_json;

Note what's absent: no ParseFromString(), no _pb2.py imports, no Python deserialization loop. The Python program registers DDL and runs SQL.

Job 2: Protobuf In, OpenSearch Out

Downstream, the sanitized Protobuf topic becomes a typed source, using the same protobuf.message-class-name property, plus ignore-parse-errors so a malformed record can't poison the pipeline.

-- SOURCE: Protobuf from the sanitized Kafka topic
CREATE TABLE kafka_source_pb (
  id STRING,
  organization_id STRING,
  event_ts ROW<`seconds` BIGINT, `nanos` INT>,
  is_active BOOLEAN,
  event_type STRING,
  kafka_timestamp TIMESTAMP(3) METADATA FROM 'timestamp'
) WITH (
  'connector' = 'kafka',
  'topic' = 'acme.events.pb.v1',
  'properties.bootstrap.servers' = 'kafka:9092',
  'scan.startup.mode' = 'latest-offset',
  'format' = 'protobuf',
  'protobuf.message-class-name' = 'com.acme.events.v1.EventOuterClass$EnrichedEvent',
  'protobuf.ignore-parse-errors' = 'true'
);

-- SINK: OpenSearch (JSON)
CREATE TABLE opensearch_sink (
  id STRING,
  organization_id STRING,
  event_ts TIMESTAMP_LTZ(3),
  is_active BOOLEAN,
  event_type STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'opensearch-2',
  'hosts' = '${OPENSEARCH_ENDPOINT}:443',
  'index' = 'acme-events-v1',
  'format' = 'json'
);

INSERT INTO opensearch_sink
SELECT
  id,
  organization_id,
  TO_TIMESTAMP_LTZ(event_ts.seconds * 1000, 3),
  is_active,
  event_type
FROM kafka_source_pb;

The Build Step: Getting Java Classes Onto Flink's Classpath

The one genuinely new piece of workflow is compiling your .proto definitions to Java and packaging them into the job's fat JAR. The essential Maven pieces:

<dependencies>
  <dependency>
    <groupId>com.google.protobuf</groupId>
    <artifactId>protobuf-java</artifactId>
    <version>3.25.5</version>
  </dependency>
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-protobuf</artifactId>
    <version>${flink.version}</version>
  </dependency>
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka</artifactId>
    <version>${flink.connector.kafka.version}</version>
  </dependency>
  <!-- plus your sink connectors, e.g. flink-connector-opensearch2 -->
</dependencies>

Two practices that made this maintainable for us:

• Version-control the generated Java sources (or generate them in CI from a single canonical .proto repo) and pull them in with build-helper-maven-plugin's add-source, rather than compiling .proto files in every consuming project. One schema source of truth, many consumers.
• Shade everything into one JAR with maven-shade-plugin, excluding signature files (META-INF/*.SF, *.DSA, *.RSA). On AWS Managed Flink, pass it via the job's JAR configuration; on self-managed Flink, drop it in lib/ or use --classpath.

The full workflow: define the .proto, compile it to Java with protoc, package the fat JAR, put it on Flink's classpath, author the PyFlink job with the DDL above, then deploy and watch end-to-end p99.

How We Measured the Improvement

We measured end-to-end p99 latency as the time from a record landing on the source Kafka topic to the corresponding OpenSearch write being acknowledged

Results

• End-to-end p99 latency around 500 milliseconds in our environment at production load, down from a 3-5 second baseline, by eliminating per-record JVM-to-Python crossings and Python-side parsing on the hot path
• Less code. The deserialization UDFs, the _pb2 imports, and their error handling all disappeared. What remains is DDL plus SQL
• Simpler and easier to operate. The pipeline now relies on Flink's Kafka connector and Protobuf format for serialization and parsing, with built-in parse-error handling, instead of hand-rolled Python parsing

When This Optimization Won't Help

Moving Protobuf decoding to the JVM won't automatically solve every latency problem. If your pipeline's critical path is dominated by sink backpressure, network latency, external API calls, state access, or checkpointing overhead rather than deserialization, changing the serialization path may have little effect on end-to-end latency. This optimization is most valuable when profiling specifically shows that Python execution and JVM/Python data movement are significant contributors to the critical path, which is why we'd recommend profiling first rather than applying this as a default change.

When You Should Still Use Python UDFs

This pattern is not "never write Python UDFs." It's "keep them off the per-record deserialization path." Python remains the right tool when:

• The transformation genuinely needs Python libraries (ML feature computation, model inference, specialized parsing that has no SQL equivalent).
• Throughput is modest and developer velocity matters more than the last hundred milliseconds.
• You're prototyping. Even then, declare the format natively from day one anyway; it costs nothing and you won't have to migrate later.

If a UDF is unavoidable on a hot path, at least let the JVM do the deserialization first so the UDF receives typed columns rather than raw bytes.

Gotchas Worth Knowing Before You Ship

• Property syntax varies by Flink version. Some versions use format = 'protobuf'; newer key/value descriptors prefer value.format = 'protobuf'. Check your version's docs.
• Enums: surface them as STRING if you need ergonomic SQL manipulation, or keep them numeric with a lookup table.
• Schema evolution: favor backward-compatible, additive changes with defaults. Because the compiled Java classes are baked into the JAR, a schema change means a rebuild and redeploy, so make that a deliberate, versioned step in CI rather than an afterthought. ignore-parse-errors is your safety net during rollout windows, but monitor the drop counter so it doesn't silently eat data.
• Benchmark end-to-end, not just the UDF: source lag, operator latency, and sink acknowledgments under production load patterns. Deserialization wins can be masked, or dwarfed, by sink backpressure.
• Security: lock down OpenSearch credentials and TLS; pin Kafka client versions compatible with your Flink release.

Closing Thoughts

We didn't rewrite the pipeline in Java. We removed an unnecessary per-record JVM-to-Python boundary from the hot path and let Flink's JVM-native Protobuf format do the work it was designed to do. If your PyFlink job parses Protobuf messages in Python today, check whether Flink's native format support can move that work into the JVM-side execution path. For latency-sensitive pipelines, eliminating unnecessary Python boundaries may be one of the highest-leverage optimizations to investigate, especially when profiling shows that serialization and Python execution are on the critical path.