Apache Flink中两类执行环境的区别及Kafka流作业选型咨询

Hey there! Let’s clear up the confusion between StreamExecutionEnvironment and StreamTableEnvironment first, then figure out which one is right for your Kafka streaming job.

Core Purpose & Programming Model #

• StreamExecutionEnvironment: This is the entry point for Flink’s DataStream API—the imperative, low-level stream processing toolkit. You’ll use it when you need full, fine-grained control over every step of your pipeline: think custom transformation functions, complex stateful operations, hand-tuned window logic, or integrating with niche connectors directly. It’s all about writing code that explicitly defines how data flows and transforms, step by step.
• StreamTableEnvironment: This is the entry point for Flink’s Table & SQL API—a declarative, high-level interface. Instead of coding each transformation, you define what you want to achieve (e.g., "filter transactions over $100", "join this stream with a user lookup table") using SQL queries or Table API methods. Flink’s optimizer handles translating your declarative logic into optimized execution plans, which cuts down on boilerplate and leverages built-in performance tweaks.

Data Representation #

• With StreamExecutionEnvironment, data lives as DataStream<T> where T is a concrete object (like a POJO, Tuple, or raw string). You work directly with these objects in your code.
• With StreamTableEnvironment, data is represented as a schema-aware Table (or a DataStream converted to a Table). This aligns with SQL’s tabular model, so you’ll need to define (or infer) a schema for your data to work with it.

Typical Use Cases #

Choose StreamExecutionEnvironment if:

• You need full control over state management, windowing, or custom serialization
• You’re dealing with unstructured/semi-structured data that doesn’t fit a strict tabular schema
• You’re building custom operators that can’t be easily expressed with SQL

Choose StreamTableEnvironment if:

• You prefer writing SQL (familiar to most data engineers) over imperative code
• You’re working with structured data and want to leverage Flink’s built-in optimizations (like predicate pushdown, join reordering)
• You need to integrate with catalogs, external tables, or use high-level time-window logic with SQL syntax

It depends on what your pipeline needs to do after consuming from Kafka:

1. If your logic is simple and structured #

For example, filtering records, aggregating by key over time windows, joining with a static lookup table, or writing to a structured sink (like JDBC or Elasticsearch with a defined schema). Go with StreamTableEnvironment—it’s concise and leverages Flink’s optimizations. Here’s a quick snippet:

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

// Define Kafka source as a table using SQL DDL
tableEnv.executeSql("""
    CREATE TABLE kafka_source (
        user_id STRING,
        purchase_amount INT,
        event_time TIMESTAMP(3) METADATA FROM 'timestamp'
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'user-purchases',
        'properties.bootstrap.servers' = 'localhost:9092',
        'format' = 'json'
    )
""");

// Run an aggregation query and write to a sink
tableEnv.executeSql("""
    INSERT INTO daily_purchase_totals
    SELECT user_id, SUM(purchase_amount) as total_spent
    FROM kafka_source
    GROUP BY user_id, TUMBLE(event_time, INTERVAL '1' DAY)
""");

2. If your logic requires custom, complex processing #

For example, parsing unstructured Kafka messages with custom logic, implementing session-based user tracking with custom state, or integrating with non-standard sinks/sources that lack Table API connectors. Go with StreamExecutionEnvironment—you’ll have full control over every step:

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// Consume raw strings from Kafka
Properties kafkaProps = new Properties();
kafkaProps.setProperty("bootstrap.servers", "localhost:9092");
DataStream<String> kafkaRawStream = env.addSource(
    new FlinkKafkaConsumer<>("user-events", new SimpleStringSchema(), kafkaProps)
);

// Custom parsing and stateful processing
DataStream<UserSession> sessionStream = kafkaRawStream
    .map(new CustomEventParser()) // Convert raw string to UserEvent POJO
    .keyBy(UserEvent::getUserId)
    .window(EventTimeSessionWindows.withGap(Time.minutes(10)))
    .aggregate(new SessionAggregateFunction()); // Custom aggregation logic

// Write to a custom sink
sessionStream.addSink(new SessionDatabaseSink());

env.execute("Kafka User Session Job");

Bonus: You can mix both! #

If your pipeline has parts that need custom processing and parts that are better suited for SQL, you can convert between DataStream and Table using StreamTableEnvironment. This lets you get the best of both worlds.

内容的提问来源于stack exchange，提问作者Yves