腾讯云流计算Oceanus对接使用全攻略:从数据接入到实时计算实战
引言:实时计算的时代选择
在数字化转型的浪潮中,企业对数据实时性的追求从未如此迫切。无论是电商平台的秒级推荐、金融风控的毫秒级响应,还是物联网设备的即时监控,实时数据处理能力已经成为企业核心竞争力的重要组成部分。腾讯云流计算Oceanus,作为基于Apache Flink构建的企业级实时大数据分析平台,正以其一站开发、无缝连接、亚秒延时、安全稳定等特性,成为众多企业实时化建设的首选利器。
然而,对于许多初次接触Oceanus的开发者和架构师来说,如何正确地将Oceanus与各类数据源和数据目的进行对接,如何高效地开发和管理实时计算作业,仍然是一个需要系统学习的课题。本文将从一个实践者的视角出发,全面深入地讲解腾讯云流计算Oceanus的对接使用方法,涵盖从环境准备、数据接入、作业开发到监控运维的全链路内容,并配以大量的代码示例和实战案例,力求让读者能够快速上手并应用于实际生产环境。
需要先登录腾讯云控制台,点击:腾讯云控制台,还没有账号,点击:注册后再关联,已有账号点击:登录后再关联
一、流计算Oceanus概述
1.1 什么是Oceanus
流计算Oceanus是腾讯云大数据产品生态体系中的实时化分析利器,它基于Apache Flink构建,具备一站式开发、无缝连接、亚秒级延时、低廉成本和安全稳定等特点。Oceanus的目标是帮助企业加速实时化数字化建设进程,实现数据价值最大化。
与开源Apache Flink相比,Oceanus在多个维度进行了增强和优化:
- 数据连接:依托丰富的Connector连接多种上下游数据服务,与腾讯云主流数据产品(如TCHouse、Ckafka、CDB、COS、ES等)无缝集成
- 开发调试:提供WebIDE一站式开发调试平台,支持SQL/Java/Scala多语言开发,具备图形化实时ETL开发能力
- 元数据管理:内置统一元数据管理,支持智能感知外部数据连接系统,自动生成数据源和数据目的建表语句
- 作业运维:支持作业版本管理、可视化运行监控、65+项监控指标、智能诊断等企业级运维能力
1.2 Oceanus的作业类型
Oceanus支持四种类型的作业,满足不同场景的开发需求:
- SQL作业:通过Flink SQL进行流数据处理,适合大多数实时ETL、数据清洗、聚合分析等场景
- JAR作业:上传自定义的Flink JAR包,适合需要复杂业务逻辑或自定义算子的场景
- Python作业:使用Python语言开发Flink作业,适合数据科学团队快速迭代
- ETL作业:图形化低代码ETL开发(注意:Oceanus已不再支持ETL作业类型的新建,存量作业可继续运行,推荐使用SQL作业或WeData实时集成)
1.3 Oceanus的部署模式
Oceanus采用的是独享集群模式,用户可以在自己的集群中运行各类作业,并进行相关的资源管理。每个集群都是独立部署的,资源隔离性良好,适合对数据安全和性能有较高要求的企业场景。
二、环境准备与集群创建
2.1 前置条件
在使用Oceanus之前,需要确保以下几点:
- 已开通腾讯云账号并完成实名认证
- 已开通Oceanus服务
- 准备好VPC私有网络(如果尚未创建,需要先创建)
- 准备好日志服务CLS(用于存储作业日志)
- 准备好对象存储COS(用于存储作业相关文件)
2.2 创建VPC私有网络
私有网络VPC是腾讯云上逻辑隔离的网络空间。在构建Oceanus集群以及后续的上下游数据服务(如MySQL、Kafka、ES等)时,必须保持网络一致才能互通。
创建VPC的步骤如下:
- 登录私有网络控制台
- 选择"私有网络" > "+新建"
- 填写VPC名称和CIDR网段
- 创建子网并选择可用区
重要提示:Oceanus集群与数据源/数据目的集群必须在同一个VPC下,否则需要通过对等连接、NAT网关等方式手动打通网络。
2.3 创建Oceanus集群
在流计算Oceanus控制台中创建独享集群:
- 登录流计算Oceanus控制台
- 进入"计算资源" > "新建"
- 选择地域和可用区(需与后续使用的其他云服务在同一地域)
- 选择VPC和子网(与上下游服务保持一致)
- 选择日志服务CLS和对象存储COS
- 设置集群初始密码
- 确认配置并完成购买
集群创建完成后,可以在集群列表中查看集群状态。通常需要等待几分钟集群才能进入运行状态。
三、数据源对接:将数据接入Oceanus
3.1 数据源概述
数据源(Source)是输入流计算系统的上游数据来源。在Oceanus SQL作业中,数据源可以是消息队列Kafka、数据库MySQL、Elasticsearch等。Oceanus内置了50+官方Connector,覆盖了主流的数据源类型:
- 消息队列:CKafka、TDMQ、RocketMQ、Pulsar
- 数据库:MySQL、PostgreSQL、TDSQL、MongoDB、Redis、Oracle、SQL Server
- 数据湖/数仓:Iceberg、Hudi、ClickHouse、DLC、TBDS
- SaaS/API:微信、企业微信、飞书、COS、CLS、ES、Grafana
3.2 对接Kafka数据源
Kafka是最常用的流数据源之一。以下是通过Flink SQL在Oceanus中创建Kafka Source的示例:
-- 创建Kafka数据源表
CREATE TABLE kafka_source (
`user_id` BIGINT,
`event_type` STRING,
`event_time` TIMESTAMP(3),
`properties` MAP<STRING, STRING>,
WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'user_events',
'properties.bootstrap.servers' = 'ckafka-xxxxx.ckafka.tencentcloud.com:9092',
'properties.group.id' = 'oceanus_consumer_group',
'scan.startup.mode' = 'latest-offset',
'format' = 'json'
);关键参数说明:
connector:指定连接器类型为kafkatopic:要消费的Kafka主题名称properties.bootstrap.servers:CKafka实例的接入地址properties.group.id:消费者组IDscan.startup.mode:消费起始位置,可选earliest-offset、latest-offset、specific-offsets、timestampformat:消息格式,支持json、csv、avro、canal-json、debezium-json等
3.3 对接MySQL CDC数据源
MySQL CDC(Change Data Capture)是实时捕获MySQL数据库变更数据的技术,常用于实时数据同步和实时数仓构建。
使用MySQL CDC前,需要对MySQL进行配置:
- 修改MySQL参数:
binlog_row_image = FULL - 确保MySQL开启了binlog,且格式为ROW
- 创建具有SELECT、REPLICATION SLAVE、REPLICATION CLIENT权限的数据库账号
创建MySQL CDC Source的SQL示例:
-- 创建MySQL CDC数据源表
CREATE TABLE mysql_cdc_source (
`id` INT,
`name` STRING,
`age` INT,
`create_time` TIMESTAMP(3),
PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc',
'hostname' = 'your-mysql-host.tencentcloud.com',
'port' = '3306',
'username' = 'cdc_user',
'password' = 'your_password',
'database-name' = 'test_db',
'table-name' = 'student',
'server-id' = '5401-5404',
'scan.incremental.snapshot.enabled' = 'true'
);关键参数说明:
connector:必须为mysql-cdchostname:MySQL数据库地址database-name和table-name:指定要监听的数据库和表server-id:MySQL CDC的server id,每个CDC连接必须唯一scan.incremental.snapshot.enabled:启用增量快照,支持并行读取
3.4 对接COS对象存储数据源
Oceanus可以读取COS中存储的文件数据作为数据源。以下是读取COS中JSON格式文件的示例:
-- 创建COS文件数据源表
CREATE TABLE cos_source (
`id` BIGINT,
`name` STRING,
`value` DOUBLE
) WITH (
'connector' = 'filesystem',
'path' = 'cosn://your-bucket-1234567890/path/to/data/',
'format' = 'json',
'source.monitor-interval' = '60s',
'fs.cosn.userinfo.secretId' = 'your_secret_id',
'fs.cosn.userinfo.secretKey' = 'your_secret_key',
'fs.cosn.bucket.region' = 'ap-guangzhou',
'fs.cosn.impl' = 'org.apache.hadoop.fs.CosFileSystem',
'fs.AbstractFileSystem.cosn.impl' = 'org.apache.hadoop.fs.CosN'
);3.5 对接CLS日志服务数据源
Oceanus可以消费腾讯云日志服务CLS中的日志数据。使用前需要上传flink-connector-cls的JAR包到Oceanus的依赖管理中。
-- 创建CLS数据源表(需先上传CLS Connector)
CREATE TABLE cls_source (
`log_time` TIMESTAMP(3),
`level` STRING,
`message` STRING,
`app_name` STRING
) WITH (
'connector' = 'cls',
'topicId' = 'your_cls_topic_id',
'logsetId' = 'your_cls_logset_id',
'secretId' = 'your_secret_id',
'secretKey' = 'your_secret_key',
'region' = 'ap-guangzhou',
'startTime' = '2026-01-01 00:00:00'
);四、数据目的对接:将计算结果输出
4.1 数据目的概述
数据目的(Sink)是流计算系统输出处理结果的目的地。在Oceanus SQL作业中,数据目的可以是消息队列Kafka、数据库MySQL、Elasticsearch等。Oceanus同样支持用户上传自定义Connector以支持更多的数据目的。
4.2 对接Elasticsearch数据目的
Elasticsearch是常用的日志分析和搜索场景的数据存储。Oceanus支持Elasticsearch 6.x和7.x版本。以下是将数据写入Elasticsearch 7.x的示例:
-- 创建Elasticsearch 7.x数据目的表
CREATE TABLE es_sink (
`id` INT,
`name` STRING,
`score` INT,
PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
'connector' = 'elasticsearch-7',
'hosts' = 'http://10.0.0.10:9200',
'index' = 'student_scores',
'username' = 'elastic',
'password' = 'your_password',
'format' = 'json',
'sink.bulk-flush.max-actions' = '1000',
'sink.bulk-flush.max-size' = '10mb',
'sink.bulk-flush.interval' = '60s'
);注意事项:
- ES Connector版本需要与购买的ES组件版本一致
- Elasticsearch Service中无需提前创建类似表的实体
- PRIMARY KEY对应Elasticsearch中的_id
4.3 对接MySQL数据目的(JDBC Sink)
Oceanus内置了flink-connector-jdbc组件,已经包含了MySQL和PostgreSQL的驱动程序。如果需要连接Oracle等其他数据库,可以通过上传自定义程序包的方式添加JDBC Driver。
-- 创建MySQL数据目的表(JDBC Sink)
CREATE TABLE mysql_sink (
`id` INT,
`name` STRING,
`age` INT,
`update_time` TIMESTAMP(3),
PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://your-mysql-host.tencentcloud.com:3306/test_db',
'username' = 'write_user',
'password' = 'your_password',
'table-name' = 'result_table',
'sink.buffer-flush.max-rows' = '100',
'sink.buffer-flush.interval' = '5s',
'sink.max-retries' = '3'
);4.4 对接Kafka数据目的
-- 创建Kafka数据目的表
CREATE TABLE kafka_sink (
`user_id` BIGINT,
`event_type` STRING,
`processed_time` TIMESTAMP(3)
) WITH (
'connector' = 'kafka',
'topic' = 'processed_events',
'properties.bootstrap.servers' = 'ckafka-xxxxx.ckafka.tencentcloud.com:9092',
'format' = 'json',
'sink.partitioner' = 'fixed'
);4.5 对接TCHouse-C数据目的
腾讯云数据仓库TCHouse-C是云原生数据仓库,Oceanus可以将实时计算结果写入TCHouse-C进行OLAP分析。使用前需确保Oceanus集群和TCHouse-C集群在同一个VPC下。
-- 创建TCHouse-C数据目的表
CREATE TABLE tchouse_sink (
`product_id` BIGINT,
`category` STRING,
`sales_count` BIGINT,
`total_revenue` DECIMAL(18, 2),
`stat_date` DATE
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://tchouse-host.tencentcloud.com:3306/analytics_db',
'username' = 'analytics_user',
'password' = 'your_password',
'table-name' = 'daily_sales_summary'
);五、完整作业开发实战
5.1 SQL作业开发流程
在Oceanus中创建SQL作业的完整流程如下:
- 登录流计算Oceanus控制台,进入某一工作空间
- 单击左侧导航"作业管理"
- 单击"新建" > "新建作业"
- 作业类型选择"SQL作业"
- 输入作业名称,选择一个运行中的集群
- 单击确定,完成作业创建
创建完成后,在作业列表中单击作业名称进入开发调试页面,在草稿状态下进行作业开发。如果需要使用特定的Connector,需要在"作业参数"中添加相应的程序包依赖。
5.2 实战案例:实时同步MySQL数据到Elasticsearch
以下是一个完整的实战案例,演示如何将MySQL中的变更数据实时同步到Elasticsearch中。
步骤1:创建数据源表(MySQL CDC)
CREATE TABLE mysql_source (
`id` INT,
`name` STRING,
`score` INT,
`update_time` TIMESTAMP(3),
PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc',
'hostname' = 'your-mysql.tencentcloud.com',
'port' = '3306',
'username' = 'cdc_user',
'password' = 'your_password',
'database-name' = 'test',
'table-name' = 'cdc_source'
);步骤2:创建数据目的表(Elasticsearch)
CREATE TABLE es_sink (
`id` INT,
`name` STRING,
`score` INT,
PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
'connector' = 'elasticsearch-7',
'hosts' = 'http://your-es.tencentcloud.com:9200',
'index' = 'student_scores',
'username' = 'elastic',
'password' = 'your_password',
'format' = 'json',
'sink.bulk-flush.max-actions' = '500'
);步骤3:编写数据处理逻辑
-- 简单数据清洗和转换
INSERT INTO es_sink
SELECT
id,
CONCAT('Student_', name) AS name,
CASE
WHEN score >= 90 THEN 5
WHEN score >= 80 THEN 4
WHEN score >= 70 THEN 3
WHEN score >= 60 THEN 2
ELSE 1
END AS score
FROM mysql_source;步骤4:发布并运行作业
在开发调试页面完成SQL编写后,单击"发布"将草稿版本发布为正式版本,然后单击"运行"启动作业。可以通过"日志"按钮查看作业运行日志。
5.3 实战案例:实时ETL与维表关联
在实际业务中,经常需要将流数据与维表进行关联以丰富数据维度。以下示例演示了如何将MySQL流数据与HBase维表关联后写入TCHouse-C:
-- 数据源表:MySQL CDC
CREATE TABLE order_source (
`order_id` BIGINT,
`product_id` INT,
`user_id` INT,
`amount` DECIMAL(18, 2),
`order_time` TIMESTAMP(3),
PRIMARY KEY (`order_id`) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc',
'hostname' = 'your-mysql.tencentcloud.com',
'port' = '3306',
'username' = 'cdc_user',
'password' = 'your_password',
'database-name' = 'order_db',
'table-name' = 'orders'
);
-- 维表:HBase(产品信息维表)
CREATE TABLE product_dim (
`product_id` INT,
`product_name` STRING,
`category` STRING,
`price` DECIMAL(18, 2),
PRIMARY KEY (`product_id`) NOT ENFORCED
) WITH (
'connector' = 'hbase-2.2',
'table-name' = 'product_dim',
'zookeeper.quorum' = 'your-hbase-zookeeper:2181',
'lookup.cache.max-rows' = '10000',
'lookup.cache.ttl' = '10min'
);
-- 数据目的表:TCHouse-C
CREATE TABLE order_analytics_sink (
`order_id` BIGINT,
`product_name` STRING,
`category` STRING,
`amount` DECIMAL(18, 2),
`order_time` TIMESTAMP(3)
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://tchouse-host.tencentcloud.com:3306/analytics_db',
'username' = 'analytics_user',
'password' = 'your_password',
'table-name' = 'order_enriched'
);
-- 流与维表关联
INSERT INTO order_analytics_sink
SELECT
o.order_id,
p.product_name,
p.category,
o.amount,
o.order_time
FROM order_source AS o
LEFT JOIN product_dim FOR SYSTEM_TIME AS OF o.order_time AS p
ON o.product_id = p.product_id;5.4 JAR作业开发
对于需要复杂业务逻辑或自定义算子的场景,可以使用JAR作业模式。JAR作业的开发流程如下:
- 使用Maven或Gradle构建Flink项目
- 编写Flink DataStream或Table API代码
- 将项目打包为JAR文件
- 在Oceanus控制台的"依赖管理"中上传JAR包
- 创建JAR类型作业,选择已上传的JAR包和主类
- 配置作业参数,发布并运行
JAR作业的优势在于可以访问用户VPC下的所有网络可达资源,包括消息队列、数据库、API服务、CVM等。同时,通过NAT网关配置路由表,还可以访问公网地址。
六、监控与运维
6.1 监控指标
Oceanus提供了65+项监控指标,涵盖作业、JobManager、TaskManager、Task等粒度,以及容器POD的精确CPU/内存指标。监控指标以Task粒度定义动态指标,支持sum、max、min、avg等维度聚合。
6.2 接入Prometheus自定义监控
Oceanus支持将Flink内置指标和自定义业务指标通过Prometheus Pushgateway推送到Prometheus服务端,并在Grafana中进行可视化展示。
配置步骤如下:
- 在作业的"作业参数" > "高级参数"中添加Prometheus配置
- 配置metrics.reporters,添加promgateway
- 设置PushGateway的IP和端口
- 如果使用腾讯云Prometheus服务,需要配置BasicAuth认证
- 发布并重启作业,等待1分钟即可在Grafana中查看数据
配置示例:
metrics.reporters: barad,oceanus,promgateway
metrics.reporter.promgateway.host: ${Prometheus_PushGateway_IP}
metrics.reporter.promgateway.port: ${Prometheus_PushGateway_Port}
metrics.reporter.promgateway.needBasicAuth: true
metrics.reporter.promgateway.password: ${Prometheus_Token}6.3 告警配置
基于腾讯云监控,Oceanus支持为各项监控指标配置告警规则,并支持AI动态阈值告警。告警通知方式包括短信、电话、邮件、企业微信机器人等。
6.4 日志与诊断
Oceanus将作业运行时的异常日志实时投递到CLS中,支持按实例ID、进程角色、进程ID进行聚合展示和分级检索。同时提供基于异常日志的一键诊断功能,能够快速分析作业异常原因并给出建议解决方案。
七、网络与权限配置
7.1 网络互通要点
Oceanus集群与上下游数据服务的网络互通是作业正常运行的前提:
- 同VPC:最推荐的方式,Oceanus集群与数据源/数据目的集群在同一VPC下,网络天然互通
- 对等连接:如果不在同一VPC,可以通过VPC对等连接实现互通
- NAT网关:如果需要访问公网资源,可以通过NAT网关配置路由
7.2 服务委托授权
Oceanus作业需要访问消息队列、云数据库等资源时,需要进行服务委托授权。在作业开发调试页面,如果未授权会弹出授权对话框,单击"前往授权"即可完成授权。
7.3 自定义连接器管理
如果内置Connector无法满足需求,用户可以上传自定义Connector:
- 在Oceanus控制台选择"连接器管理" > "自定义连接器"
- 单击"创建自定义连接器"
- 选择依赖JAR包和版本
- 等待Oceanus解析完成
- 选择对应的Connector类型,单击确定
八、成本优化建议
8.1 资源规格选择
Oceanus采用按量计费模式,合理选择资源规格可以有效控制成本。建议根据作业的QPS和数据量选择合适的CU(计算单元)数量,避免资源浪费或不足。
8.2 首购优惠
腾讯云为新用户提供Oceanus 5折首购优惠,最低可达0.23元/核时。相比自建Flink集群,可以节省60%以上的成本。
8.3 作业优化
- 合理设置Checkpoint间隔,避免过于频繁影响性能
- 使用状态后端(RocksDB)优化大状态作业
- 合理设置并行度,避免资源浪费
- 利用维表缓存减少外部数据源访问
九、常见问题与最佳实践
9.1 作业启动失败排查
作业启动失败时,可以通过以下步骤排查:
- 查看作业日志,定位异常堆栈
- 检查网络连通性(VPC、安全组等)
- 检查Connector版本是否匹配
- 检查数据源/数据目的配置是否正确
9.2 数据延迟问题
- 检查并行度是否充足
- 检查Checkpoint是否正常完成
- 检查背压(Backpressure)情况
- 考虑增加TaskManager资源或调整算子并行度
9.3 最佳实践总结
- 生产环境建议使用独享集群,保障资源隔离
- 开发调试阶段使用草稿版本,验证通过后再发布
- 合理配置作业参数,如Checkpoint间隔、状态后端等
- 充分利用Oceanus的监控告警能力,及时发现问题
- 定期review作业性能,进行必要的优化调整
结语
腾讯云流计算Oceanus作为企业级实时大数据分析平台,通过丰富的Connector生态、灵活的作业开发模式和完善的运维监控能力,极大地降低了实时计算的门槛和成本。本文从集群创建、数据源对接、数据目的配置、作业开发到监控运维,系统地介绍了Oceanus的对接使用方法,并提供了大量的Flink SQL代码示例和实战案例。
随着实时计算需求的不断增长,Oceanus也在持续演进,不断推出新的Connector和功能特性。建议开发者持续关注Oceanus的官方文档和版本更新,充分利用平台能力构建高效稳定的实时数据处理管道。
常见问题问答
问1:Oceanus支持哪些数据源和数据目的?
答:Oceanus内置了50+官方Connector,支持消息队列(CKafka、TDMQ等)、数据库(MySQL、PostgreSQL、Oracle、MongoDB等)、数据湖/数仓(Iceberg、ClickHouse、TCHouse-C等)、SaaS服务(COS、CLS、ES等)等多种上下游数据服务。同时支持上传自定义Connector以对接更多数据源。
问2:Oceanus集群必须与数据服务在同一个VPC吗?
答:强烈建议保持在同一VPC下,这样可以保证网络直接互通。如果不在同一VPC,需要通过VPC对等连接、NAT网关等方式手动打通网络。
问3:MySQL CDC同步需要做哪些准备工作?
答:需要修改MySQL的binlog_row_image参数为FULL,确保binlog开启且格式为ROW,并创建具有SELECT、REPLICATION SLAVE、REPLICATION CLIENT权限的数据库账号。
问4:Oceanus支持哪几种作业类型?
答:Oceanus支持SQL作业、JAR作业和Python作业三种类型。ETL作业类型已不再支持新创建,推荐使用SQL作业或WeData实时集成。
问5:如何在Oceanus中监控作业的运行状态?
答:Oceanus提供了65+项监控指标,涵盖作业、JobManager、TaskManager等粒度。同时支持接入Prometheus+Grafana进行自定义监控,以及配置告警规则通过短信、邮件、企业微信等方式发送告警。
问6:Oceanus的计费方式是什么?有没有优惠?
答:Oceanus采用按量计费模式,按使用的CU(计算单元)和时长计费。新用户可享受5折首购优惠,最低0.23元/核时。相比自建Flink集群可节省60%以上成本。




