阿里云实时计算Flink版对接实战：从数据源到结果表的全链路开发指南-阿里云开发者社区

1. 认识阿里云实时计算Flink版

阿里云实时计算Flink版是阿里云基于Apache Flink构建的企业级实时计算平台，它完全兼容开源Flink的API和生态，同时集成了阿里云强大的基础设施与服务。与自建Flink集群相比，实时计算Flink版提供了开箱即用的管控能力、弹性资源伸缩、丰富的云产品连接器以及全托管的运维服务，让开发者能够专注于业务逻辑的开发而非底层集群的搭建与维护。

实时计算Flink版的核心计算引擎是Ververica Runtime（VVR），这是阿里云对开源Flink进行深度优化和增强的版本。VVR在兼容性、性能和稳定性方面都有显著提升，尤其在处理大状态、高吞吐场景时表现优异。目前实时计算Flink版支持Flink 1.13、1.15、1.17等多个主流版本，用户可以根据业务需求灵活选择。

在应用场景上，实时计算Flink版广泛应用于实时数仓建设、实时风控、实时推荐、实时监控告警、数据清洗与ETL等领域。无论是毫秒级的实时计算还是海量数据的流式处理，实时计算Flink版都能提供稳定可靠的计算能力。

需要先登录阿里云控制台，点击：阿里云控制台

2. 实时计算Flink版的核心概念

在使用实时计算Flink版之前，有必要了解几个核心概念：

2.1 工作空间

工作空间是实时计算Flink版的基本资源单元，类似于一个独立的Flink集群环境。每个工作空间都包含独立的计算资源（CPU和内存）、存储资源和网络配置。用户可以在工作空间内创建、提交和管理Flink作业。工作空间分为共享模式和独享模式两种，共享模式适合开发和测试场景，独享模式则适合对性能和隔离性有较高要求的生产环境。

2.2 作业

作业是实时计算Flink版中执行的具体计算任务，对应一个Flink应用程序。每个作业都有自己的执行图（Execution Graph）、状态后端（State Backend）和并行度配置。作业的生命周期包括：草稿、启动中、运行中、停止中、已停止、失败等状态。

2.3 部署模式

实时计算Flink版支持两种主要的作业部署模式：

Flink SQL模式：通过编写SQL语句来定义数据处理逻辑，适合数据清洗、聚合、维表关联等常见场景，开发效率高，门槛低。
DataStream模式：通过Java或Scala编写Flink DataStream API程序，适合需要复杂状态管理、自定义算子、事件时间处理等高级场景，灵活性更强。

2.4 连接器

连接器（Connector）是Flink与外部数据源/数据目的地进行交互的组件。实时计算Flink版内置了丰富的连接器，包括Kafka、DataHub、Hologres、RDS MySQL、Redis、OTS（表格存储）、Elasticsearch、HBase等，覆盖了绝大多数实时数据处理场景。

3. 准备工作：创建工作空间与获取访问信息

在使用实时计算Flink版之前，首先需要完成以下准备工作：

3.1 开通实时计算Flink版服务

登录阿里云控制台，在产品列表中找到“实时计算Flink版”，点击开通服务。首次开通时需要进行实名认证和费用充值，实时计算Flink版支持按量付费和包年包月两种计费模式。按量付费适合短期或波动性较大的项目，包年包月则适合长期稳定运行的生产环境。

3.2 创建工作空间

开通服务后，进入实时计算Flink版控制台，点击“创建工作空间”。在创建过程中需要选择工作空间的名称、地域（Region）、可用区、计算资源规格（如cu数）以及存储资源（用于存放作业状态和Checkpoint）。

3.3 获取VPC和交换机信息

实时计算Flink版作业需要访问VPC内的其他云服务（如RDS、Kafka等），因此需要在创建时指定VPC和交换机。如果还没有VPC，可以提前在专有网络控制台创建。获取VPC ID、交换机ID和可用区信息后，在创建工作空间时填入。

3.4 创建访问密钥

如果作业需要读写OSS、DataHub等云服务，需要准备AccessKey ID和AccessKey Secret。建议使用RAM子账号的AccessKey，并遵循最小权限原则，只授予必要的权限。

4. 数据源对接：从Kafka读取实时数据

Kafka是实时计算中最常用的数据源之一。下面以Flink SQL为例，演示如何创建Kafka数据源表并读取数据。

4.1 创建Kafka数据源表（Flink SQL）

CREATE TEMPORARY TABLE kafka_source (
    `user_id` BIGINT,
    `item_id` BIGINT,
    `action` STRING,
    `action_time` TIMESTAMP(3),
    WATERMARK FOR action_time AS action_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'user_action_log',
    'properties.bootstrap.servers' = '172.16.0.10:9092,172.16.0.11:9092',
    'properties.group.id' = 'flink_consumer_group',
    'scan.startup.mode' = 'latest-offset',
    'format' = 'json',
    'json.ignore-parse-errors' = 'true'
);

上述代码创建了一个名为kafka_source的临时表，用于从Kafka主题user_action_log中读取JSON格式的数据。关键参数说明：

connector：指定连接器类型为kafka。
topic：要消费的Kafka主题名称。
properties.bootstrap.servers：Kafka集群的地址列表。
properties.group.id：消费者组ID，用于offset管理。
scan.startup.mode：消费起始位置，支持earliest-offset（最早）、latest-offset（最新）、timestamp（指定时间戳）等。
format：消息格式，支持json、csv、avro、canal-json、debezium-json等。

4.2 对接阿里云DataHub

DataHub是阿里云提供的流式数据总线服务，与实时计算Flink版有很好的集成。创建DataHub数据源表的SQL如下：

CREATE TEMPORARY TABLE datahub_source (
    `order_id` BIGINT,
    `user_id` BIGINT,
    `amount` DECIMAL(10, 2),
    `order_time` TIMESTAMP(3)
) WITH (
    'connector' = 'datahub',
    'endpoint' = 'http://dh-cn-hangzhou.aliyuncs.com',
    'project' = 'your_project_name',
    'topic' = 'your_topic_name',
    'accessId' = 'your_access_key_id',
    'accessKey' = 'your_access_key_secret',
    'startTime' = '2026-01-01 00:00:00'
);

与Kafka不同，DataHub连接器需要指定endpoint、project和topic，以及用于鉴权的accessId和accessKey。startTime参数用于指定消费的起始时间点。

5. 维表对接：实时关联MySQL维表数据

在实时计算中，经常需要将流数据与静态或缓慢变化的维度数据进行关联，例如将用户行为日志与用户信息表关联。实时计算Flink版支持将MySQL、Redis等作为维表进行Lookup Join。

5.1 MySQL维表配置（Flink SQL）

CREATE TEMPORARY TABLE dim_user (
    `user_id` BIGINT,
    `user_name` STRING,
    `user_level` INT,
    `register_time` TIMESTAMP(3),
    PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://172.16.0.20:3306/flink_dim',
    'table-name' = 'user_info',
    'username' = 'flink_user',
    'password' = 'your_password',
    'lookup.cache.max-rows' = '10000',
    'lookup.cache.ttl' = '60000'
);

上述代码创建了一个名为dim_user的维表，它对应MySQL数据库中的user_info表。关键参数说明：

lookup.cache.max-rows：维表缓存的最大行数，用于减少对数据库的查询压力。
lookup.cache.ttl：缓存的有效时间（毫秒），超过该时间后缓存失效，需要重新查询。

启用缓存可以显著提升维表关联的性能，但需要注意缓存的时效性。对于数据变化频繁的维表，可以适当缩短TTL或禁用缓存。

5.2 使用维表进行JOIN

创建好维表后，可以通过普通JOIN语法将流数据与维表关联：

SELECT
    s.user_id,
    d.user_name,
    d.user_level,
    s.item_id,
    s.action,
    s.action_time
FROM kafka_source s
LEFT JOIN dim_user FOR SYSTEM_TIME AS OF s.action_time AS d
ON s.user_id = d.user_id;

这里使用了FOR SYSTEM_TIME AS OF语法，表示在查询时使用维表在流数据事件时间的最新快照。这是Flink SQL中标准的维表JOIN写法。

5.3 Redis维表配置

对于需要更高吞吐量的维表查询场景，Redis是比MySQL更好的选择。Redis维表的配置如下：

CREATE TEMPORARY TABLE dim_redis (
    `key` STRING,
    `value` STRING
) WITH (
    'connector' = 'redis',
    'host' = '172.16.0.30',
    'port' = '6379',
    'password' = 'your_redis_password',
    'command' = 'GET'
);

Redis维表通常以key-value形式存储，在JOIN时通过key来查找对应的value。

6. 结果表对接：将计算结果写入目标存储

实时计算的结果需要写入到下游存储系统中，供业务系统查询或进一步处理。实时计算Flink版支持多种结果表类型。

6.1 写入Hologres

Hologres是阿里云自主研发的交互式分析引擎，支持实时写入和实时查询，是实时数仓的最佳搭档。将结果写入Hologres的SQL如下：

CREATE TEMPORARY TABLE hologres_sink (
    `user_id` BIGINT,
    `item_id` BIGINT,
    `action` STRING,
    `cnt` BIGINT,
    `window_start` TIMESTAMP(3),
    `window_end` TIMESTAMP(3)
) WITH (
    'connector' = 'hologres',
    'endpoint' = 'hgpostcn-cn-hangzhou-internal.hologres.aliyuncs.com:80',
    'dbname' = 'your_db_name',
    'tablename' = 'your_table_name',
    'username' = 'your_username',
    'password' = 'your_password',
    'batchSize' = '1000',
    'sink.parallelism' = '4'
);

Hologres连接器支持批量写入和流式写入两种模式。batchSize参数控制批量写入的大小，适当调大可以提升写入吞吐量。sink.parallelism控制写入任务的并行度。

6.2 写入RDS MySQL

对于需要将计算结果存储到关系型数据库的场景，可以使用JDBC连接器写入MySQL：

CREATE TEMPORARY TABLE mysql_sink (
    `user_id` BIGINT,
    `total_amount` DECIMAL(10, 2),
    `order_count` BIGINT,
    `stat_date` STRING
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://172.16.0.20:3306/flink_result',
    'table-name' = 'user_stat',
    'username' = 'flink_user',
    'password' = 'your_password',
    'sink.buffer-flush.max-rows' = '500',
    'sink.buffer-flush.interval' = '5000'
);

sink.buffer-flush.max-rows和sink.buffer-flush.interval分别控制缓冲区刷新行数和刷新间隔，合理配置可以平衡写入性能和实时性。

6.3 写入OTS（表格存储）

OTS是阿里云提供的NoSQL数据库，适合海量数据的存储和查询：

CREATE TEMPORARY TABLE ots_sink (
    `pk` STRING,
    `col_user_id` BIGINT,
    `col_item_id` BIGINT,
    `col_action` STRING,
    PRIMARY KEY (pk) NOT ENFORCED
) WITH (
    'connector' = 'ots',
    'endpoint' = 'https://your-instance.cn-hangzhou.ots.aliyuncs.com',
    'instance' = 'your_instance_name',
    'table' = 'your_table_name',
    'accessId' = 'your_access_key_id',
    'accessKey' = 'your_access_key_secret'
);

7. 完整作业示例：实时UV统计与维表关联

下面通过一个完整的Flink SQL作业示例，展示从数据源读取、维表关联、窗口聚合到结果写入的全流程。

7.1 业务场景

假设有一个用户行为日志流（来自Kafka），需要实时统计每个商品（item_id）在过去5分钟内的独立访客数（UV），同时关联用户维表获取用户等级信息，并将结果写入Hologres供前端展示。

7.2 完整SQL代码

-- 1. 创建Kafka数据源表
CREATE TEMPORARY TABLE kafka_source (
    `user_id` BIGINT,
    `item_id` BIGINT,
    `action` STRING,
    `action_time` TIMESTAMP(3),
    WATERMARK FOR action_time AS action_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'user_action_log',
    'properties.bootstrap.servers' = '172.16.0.10:9092',
    'properties.group.id' = 'flink_uv_group',
    'scan.startup.mode' = 'latest-offset',
    'format' = 'json'
);
-- 2. 创建MySQL用户维表
CREATE TEMPORARY TABLE dim_user (
    `user_id` BIGINT,
    `user_level` INT,
    PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://172.16.0.20:3306/flink_dim',
    'table-name' = 'user_info',
    'username' = 'flink_user',
    'password' = 'your_password',
    'lookup.cache.max-rows' = '5000',
    'lookup.cache.ttl' = '30000'
);
-- 3. 创建Hologres结果表
CREATE TEMPORARY TABLE hologres_sink (
    `item_id` BIGINT,
    `uv_count` BIGINT,
    `avg_user_level` DOUBLE,
    `window_start` TIMESTAMP(3),
    `window_end` TIMESTAMP(3)
) WITH (
    'connector' = 'hologres',
    'endpoint' = 'hgpostcn-cn-hangzhou-internal.hologres.aliyuncs.com:80',
    'dbname' = 'flink_realtime',
    'tablename' = 'item_uv_stat',
    'username' = 'your_username',
    'password' = 'your_password',
    'batchSize' = '500'
);
-- 4. 执行计算并写入结果
INSERT INTO hologres_sink
SELECT
    s.item_id,
    COUNT(DISTINCT s.user_id) AS uv_count,
    AVG(CAST(d.user_level AS DOUBLE)) AS avg_user_level,
    TUMBLE_START(s.action_time, INTERVAL '5' MINUTE) AS window_start,
    TUMBLE_END(s.action_time, INTERVAL '5' MINUTE) AS window_end
FROM kafka_source s
LEFT JOIN dim_user FOR SYSTEM_TIME AS OF s.action_time AS d
ON s.user_id = d.user_id
GROUP BY
    s.item_id,
    TUMBLE(s.action_time, INTERVAL '5' MINUTE);

7.3 作业提交与运行

编写好SQL后，可以通过实时计算Flink版控制台的“SQL开发”页面提交作业。在提交前，需要配置作业的并行度、Checkpoint间隔、状态后端等参数。建议在生产环境中开启Checkpoint，并设置合理的间隔（如60秒），以保证作业的容错性。

8. DataStream模式开发与对接

对于需要更精细控制计算逻辑的场景，可以使用DataStream API进行开发。实时计算Flink版完全兼容开源Flink的DataStream API，开发者可以将本地开发的JAR包上传并提交到工作空间运行。

8.1 Maven依赖配置

在项目的pom.xml中添加实时计算Flink版的相关依赖：

<dependency>
    <groupId>com.alibaba.ververica</groupId>
    <artifactId>ververica-connector-kafka</artifactId>
    <version>1.15-vvr-6.0.2</version>
</dependency>
<dependency>
    <groupId>com.alibaba.ververica</groupId>
    <artifactId>ververica-connector-hologres</artifactId>
    <version>1.15-vvr-6.0.2</version>
</dependency>
<dependency>
    <groupId>com.alibaba.ververica</groupId>
    <artifactId>ververica-connector-jdbc</artifactId>
    <version>1.15-vvr-6.0.2</version>
</dependency>

注意版本号需要与工作空间使用的Flink版本相匹配。

8.2 Kafka数据源（DataStream）

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class KafkaSourceExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
            .setBootstrapServers("172.16.0.10:9092")
            .setTopics("user_action_log")
            .setGroupId("flink_consumer_group")
            .setStartingOffsets(OffsetsInitializer.latest())
            .setValueOnlyDeserializer(new SimpleStringSchema())
            .build();
        DataStream<String> stream = env
            .fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Kafka Source");
        stream.print();
        env.execute("Kafka Source Example");
    }
}

8.3 写入Hologres（DataStream）

import com.alibaba.ververica.connectors.hologres.api.HologresConnectionPool;
import com.alibaba.ververica.connectors.hologres.api.HologresSinkFunction;
public class HologresSinkExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<String> sourceStream = ...; // 从Kafka读取的数据流
        HologresSinkFunction<String> sinkFunction = new HologresSinkFunction<>(
            "hgpostcn-cn-hangzhou-internal.hologres.aliyuncs.com:80",
            "flink_realtime",
            "item_uv_stat",
            "your_username",
            "your_password",
            null,
            1000
        );
        sourceStream.addSink(sinkFunction);
        env.execute("Hologres Sink Example");
    }
}

8.4 JAR包上传与提交

将项目打包成JAR文件后，在实时计算Flink版控制台的“作业开发”页面选择“DataStream模式”，上传JAR包并配置作业参数（如主类名、并行度、JVM参数等），然后点击提交即可运行。

9. 作业运维与监控

9.1 Checkpoint配置

Checkpoint是Flink容错机制的核心。在实时计算Flink版中，可以通过作业参数配置Checkpoint：

ExecutionConfig config = env.getConfig();
config.setCheckpointingEnabled(true);
config.setCheckpointInterval(60000); // 60秒
config.setCheckpointTimeout(120000); // 2分钟超时
config.setMinPauseBetweenCheckpoints(30000); // 最小间隔30秒
config.setMaxConcurrentCheckpoints(1);

合理的Checkpoint配置可以保证作业在发生故障时能够快速恢复，同时避免对正常处理造成过大影响。

9.2 监控指标

实时计算Flink版控制台提供了丰富的监控指标，包括：

作业吞吐量：每秒处理的数据条数。
延迟：从数据进入Flink到输出的时间差。
Checkpoint大小和耗时：用于评估状态大小和Checkpoint性能。
背压：反映作业各算子之间的处理压力。
资源使用率：CPU、内存的使用情况。

通过监控这些指标，可以及时发现作业的异常情况，并进行相应的调优。

9.3 常见报错与处理

Kafka消费超时：检查Kafka集群网络连通性，适当增加session.timeout.ms和max.poll.interval.ms参数。
维表查询超时：优化维表查询SQL，增加索引，或调整缓存配置。
状态过大导致OOM：调整状态后端为RocksDB，或增加TaskManager内存。
数据倾斜：对Key进行加盐或使用两阶段聚合。
Checkpoint失败：检查状态后端存储是否可用，适当增加Checkpoint超时时间。

10. 最佳实践与性能调优

10.1 资源规划

实时计算Flink版的资源以CU（Compute Unit）为单位，1 CU约等于1核CPU + 4GB内存。在规划资源时，需要根据数据量、计算复杂度和并行度来估算。一般建议：

Kafka分区数决定Source并行度的上限。
KeyBy操作后的算子并行度建议为Kafka分区数的整数倍。
Sink并行度可以根据目标存储的写入能力来调整。

10.2 状态管理

对于有状态的计算（如窗口聚合、去重等），状态大小直接影响到作业的性能和稳定性。建议：

使用RocksDB作为状态后端，以支持大状态场景。
合理设置状态TTL（Time-To-Live），及时清理过期状态。
使用增量Checkpoint以减少Checkpoint的耗时。

10.3 数据倾斜处理

数据倾斜是实时计算中常见的问题。处理方案包括：

对倾斜的Key进行加盐（添加随机后缀），然后分两阶段聚合。
使用Flink的MiniBatch聚合功能，在微批次内进行预聚合。
调整并行度，将热点Key分散到更多的子任务中。

11. 常见问题解答

问：实时计算Flink版和开源Flink有什么区别？

答：实时计算Flink版是基于Apache Flink构建的全托管商业产品，它提供了开箱即用的管控能力、更丰富的阿里云生态连接器、企业级的安全和稳定性保障，以及专业的技术支持服务。而开源Flink需要自行搭建和维护集群。

问：如何选择Flink SQL和DataStream两种开发模式？

答：Flink SQL适合数据清洗、聚合、维表关联等标准ETL场景，开发效率高；DataStream适合需要自定义算子、复杂状态管理、事件时间精细化控制等高级场景，灵活性更强。建议优先使用Flink SQL，在SQL无法满足需求时再考虑DataStream。

问：实时计算Flink版如何保证数据不丢失？

答：通过开启Checkpoint和配置合理的状态后端，Flink可以实现精确一次（Exactly-Once）的处理语义。同时，数据源（如Kafka）和结果表（如Hologres）也需要支持幂等写入或事务写入，才能端到端保证数据不丢失。

问：维表JOIN的性能如何优化？

答：可以从以下几个方面优化：开启维表缓存（lookup.cache.max-rows和lookup.cache.ttl）；在维表数据库上建立合适的索引；使用Redis等高性能缓存数据库作为维表；对于变化不频繁的维表，可以使用定时全量加载的方式。

问：作业出现反压（Backpressure）怎么办？

答：反压通常是由于下游处理速度跟不上上游数据产生速度导致的。可以通过增加下游算子的并行度、优化计算逻辑、增加资源等方式解决。也可以检查是否有数据倾斜或外部存储写入瓶颈。

问：如何监控实时计算Flink版作业的运行状态？

答：可以通过实时计算Flink版控制台查看作业的运行状态、吞吐量、延迟、Checkpoint等指标。也可以配置云监控告警，在作业失败或指标异常时及时收到通知。

探索云世界

热门

云计算

大数据

云原生

人工智能

数据库

开发与运维

活动广场

任务中心

训练营

直播

乘风者计划

下载

镜像站

技术资料

阿里云实时计算Flink版对接实战：从数据源到结果表的全链路开发指南