腾讯云流计算Oceanus对接使用完全指南：从零搭建实时数据处理管道

apphuang2026年07月10日 10:11:4910

1. 流计算Oceanus产品概述

流计算Oceanus是腾讯云大数据产品生态体系中的实时化分析利器，基于Apache Flink构建，具备一站开发、无缝连接、亚秒延时、低廉成本、安全稳定等特点。作为全托管的云上服务，用户无需关注基础设施的运维，便能便捷对接云上数据源，获得完善的配套支持。Oceanus提供全托管的独享集群模式，用户可以在自己的集群中运行各类作业，并进行相关资源管理。计费以CU为最小单位，1CU的具体含义为CPU 1核、内存4GB。

Oceanus支持四种作业类型：SQL作业、JAR作业、Python作业和ETL作业。其中ETL作业类型已不再支持新建，推荐使用SQL作业或WeData实时集成功能完成ETL业务。Oceanus内置了30多种上下游Connector，覆盖消息队列、数据库、数据仓库、数据湖等多种数据源。

需要先登录腾讯云控制台，点击：腾讯云控制台，还没有账号，点击：注册后再关联，已有账号点击：登录后再关联

2. 环境准备与集群创建

2.1 创建独享集群

登录流计算Oceanus控制台，在计算资源页面选择地域，单击新建创建独享集群。创建集群时需要配置以下关键参数：

  • 地域与可用区：集群需与后续使用的其他云资源（如CKafka、云数据库等）处于同一地域和可用区，否则需通过对等连接等方式打通网络
  • 私有网络VPC与子网：VPC及子网需与MySQL、ES等组件使用同一个，否则需要手动打通
  • 日志配置：可选择CLS日志服务或COS存储作业运行日志
  • 存储配置：用于存放作业状态快照等数据

集群创建成功后，单击操作中的关联空间，将集群关联到某一工作空间，关联后该工作空间即可使用该集群的计算资源运行作业。

2.2 网络连通性要点

私有网络VPC是一块在腾讯云上自定义的逻辑隔离网络空间。Oceanus集群与数据源、数据汇之间的网络连通是作业正常运行的前提。最佳实践是让Oceanus集群与所有上下游服务部署在同一个VPC下。若无法满足，则需要通过对等连接、NAT网关或VPN等方式打通网络。

3. 作业创建与开发

3.1 创建作业

进入工作空间后，在作业管理页面单击新建>新建作业。选择作业类型（SQL/JAR/Python），输入作业名称，选择运行中的集群，单击确定即可完成创建。创建完成后，在作业列表中单击作业名称进入开发调试页面。

3.2 SQL作业开发

SQL作业是Oceanus最常用的作业类型，支持标准的Flink SQL语法。开发流程如下：

  1. 服务委托授权：首次开发需单击前往授权，授权流计算作业访问消息队列或云数据库等资源
  2. 编写SQL语句：在代码编辑框中编写Source、Sink定义及业务逻辑
  3. 添加Connector依赖：在作业参数中选择所需的内置Connector，如mysql-cdc connector、elasticsearch connector等
  4. 设置作业参数：配置Checkpoint间隔、算子并行度等
  5. 发布运行：单击发布草稿，检查通过后生成新版本，单击运行版本启动作业

以下是一个使用datagen数据源和blackhole数据目的的基础示例：

-- 创建数据源表（Source），datagen为内置随机数据生成器
CREATE TABLE Data_Input (
    age BIGINT,
    score BIGINT
) WITH (
    'connector' = 'datagen',
    'rows-per-second' = '100',
    'fields.age.kind' = 'random',
    'fields.age.min' = '1',
    'fields.age.max' = '100',
    'fields.score.kind' = 'random',
    'fields.score.min' = '1',
    'fields.score.max' = '1000'
);

-- 创建数据结果表（Sink），blackhole为本地数据黑洞
CREATE TABLE Data_Output (
    avg_age BIGINT,
    avg_score BIGINT
) WITH (
    'connector' = 'blackhole'
);

-- 业务逻辑：计算age和score的平均值
INSERT INTO Data_Output
SELECT AVG(age), AVG(score) FROM Data_Input;

3.3 JAR作业开发

对于复杂的自定义逻辑，可选择JAR作业类型。开发流程如下：

  1. 使用Maven或Gradle构建Flink作业代码，打包为JAR文件
  2. 在Oceanus控制台上传作业JAR包
  3. 部署并启动JAR作业

3.4 Python作业开发

Oceanus支持PyFlink，可使用Python编写流处理作业。通过一个处理数据后存入MySQL的示例可快速上手。

4. 数据源与数据汇对接

4.1 JDBC Connector（MySQL/PostgreSQL）

JDBC Connector提供了对MySQL、PostgreSQL、Oracle等常见数据库的读写支持。Oceanus已内置MySQL和PostgreSQL的驱动程序。如需连接Oracle等其他数据库，可通过附加自定义程序包的方式上传相应的JDBC Driver JAR包。

作为数据源表（Source）按列扫描：

CREATE TABLE jdbc_source_table (
    id INT,
    name STRING
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://10.1.28.93:3306/CDB?rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai',
    'table-name' = 'my-table',
    'username' = 'admin',
    'password' = 'MyPa$$w0rd',
    'scan.partition.column' = 'id',
    'scan.partition.num' = '2',
    'scan.partition.lower-bound' = '0',
    'scan.partition.upper-bound' = '100',
    'scan.fetch-size' = '1'
);

作为维表数据源（Source）：

CREATE TABLE jdbc_dim_table (
    id INT,
    name STRING
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://10.1.28.93:3306/CDB?rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai',
    'table-name' = 'my-table',
    'username' = 'admin',
    'password' = 'MyPa$$w0rd',
    'lookup.cache.max-rows' = '100',
    'lookup.cache.ttl' = '5000'
);

作为数据目的表（Tuple Sink）：

CREATE TABLE jdbc_sink_table (
    id INT,
    name STRING
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://10.1.28.93:3306/CDB?rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai',
    'table-name' = 'my-table',
    'username' = 'admin',
    'password' = 'MyPa$$w0rd'
);

4.2 MySQL CDC Connector

MySQL CDC（Change Data Capture）是实时捕获MySQL数据库变更记录的核心能力。使用CDC需要先在MySQL中修改binlog参数：将binlog_row_image设置为FULL。

以下是一个从MySQL捕获变更并输出到Elasticsearch的完整示例：

-- 创建MySQL CDC源表
CREATE TABLE mysql_cdc_source (
    id INT,
    score INT,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'your-mysql-host',
    'port' = '3306',
    'username' = 'admin',
    'password' = 'MyPa$$w0rd',
    'database-name' = 'test',
    'table-name' = 'cdc_source4es'
);

-- 创建Elasticsearch Sink表
CREATE TABLE es_sink (
    id INT,
    score INT
) WITH (
    'connector' = 'elasticsearch-7',
    'hosts' = 'http://your-es-host:9200',
    'index' = 'student_score',
    'document-type' = '_doc',
    'sink.bulk-flush.max-actions' = '1'
);

-- 业务逻辑：实时同步MySQL数据到ES
INSERT INTO es_sink SELECT id, score FROM mysql_cdc_source;

注意ES Connector版本需与购买的ES组件版本一致。Elasticsearch中无需提前创建类似表的实体。

4.3 Kafka Connector（CKafka）

Oceanus与腾讯云CKafka深度集成，支持将CKafka Topic作为数据源或数据汇。典型的数据管道为：CKafka -> Oceanus -> MySQL。

-- 创建Kafka源表
CREATE TABLE kafka_source (
    user_id STRING,
    event_time TIMESTAMP(3),
    event_data STRING,
    WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'topic_cep_demo',
    'properties.bootstrap.servers' = 'your-ckafka-broker:9092',
    'properties.group.id' = 'oceanus-consumer-group',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'json'
);

-- 创建Kafka结果表
CREATE TABLE kafka_sink (
    user_id STRING,
    processed_time TIMESTAMP(3),
    result STRING
) WITH (
    'connector' = 'kafka',
    'topic' = 'output_topic',
    'properties.bootstrap.servers' = 'your-ckafka-broker:9092',
    'format' = 'json'
);

4.4 COS对象存储Connector

Oceanus支持将数据写入COS对象存储。写入COS时，Oceanus作业所运行的地域必须和COS在同一个地域。以下示例使用datagen生成随机数据并写入COS：

-- 创建随机数据源表
CREATE TABLE random_source (
    f_sequence INT,
    f_random INT,
    f_random_str VARCHAR
) WITH (
    'connector' = 'datagen',
    'rows-per-second' = '10',
    'fields.f_sequence.kind' = 'random',
    'fields.f_sequence.min' = '1',
    'fields.f_sequence.max' = '10',
    'fields.f_random.kind' = 'random',
    'fields.f_random.min' = '1',
    'fields.f_random.max' = '100',
    'fields.f_random_str.length' = '10'
);

-- 创建COS Sink表（请替换存储桶名称和文件夹名称）
CREATE TABLE cos_sink (
    f_sequence INT,
    f_random INT,
    f_random_str VARCHAR
) PARTITIONED BY (f_sequence) WITH (
    'connector' = 'filesystem',
    'path' = 'cosn://<存储桶名称>/<文件夹名称>/',
    'format' = 'json',
    'sink.rolling-policy.file-size' = '128MB',
    'sink.rolling-policy.rollover-interval' = '30 min',
    'sink.partition-commit.delay' = '1 s',
    'sink.partition-commit.policy.kind' = 'success-file'
);

INSERT INTO cos_sink SELECT * FROM random_source;

在作业参数中需选择内置Connector flink-connector-cos，并在高级参数中配置COS地址：

fs.AbstractFileSystem.cosn.impl: org.apache.hadoop.fs.CosN
fs.cosn.impl: org.apache.hadoop.fs.CosFileSystem

4.5 CLS日志服务Connector

CLS（Cloud Log Service）既可作为Oceanus的数据源，也可作为数据目的。在集群创建时可配置CLS日志主题，作业运行期间的JobManager和TaskManager日志都会采集到CLS。

5. 性能调优

5.1 自动调优

Oceanus提供的自动调优功能可帮助用户合理调整作业并行度和资源配置，解决吞吐量不足、资源浪费等性能问题。支持SQL和JAR作业。

调优策略：

  • 负载调优策略：监控作业负载情况，根据负载自动调整资源
  • 延时调优策略（默认）：在负载调优基础上增加对消费数据源延迟的监控，延迟超过设定阈值时自动增加并行度

调优规则：

  • 自动调优会调整作业并发度以满足流量变化所需的吞吐
  • 基于延迟策略：Delay超过默认阈值60秒且持续上升3分钟，调整并发度到当前实际TPS的两倍（不超过最大资源限制）
  • 基于负载策略：某节点连续3分钟实际处理数据时间占比超过80%，调大并发度使使用率降至50%
  • 所有TaskManager CPU使用率连续24小时低于20%且处理时间低于20%时，调低并发度

开启方式：在作业管理的调优页面单击自动调优开关即可。

5.2 自定义调优（定时调优）

对于存在潮汐流量特征的作业（如直播场景夜间流量高、白天流量低），可使用自定义调优功能。支持按单次、每日、每周、每月周期配置调优计划。

注意事项：

  • 自定义调优目前为公开测试阶段，不建议对重要生产任务开启
  • 触发后需重启作业，大状态作业可能导致长时间停流
  • 开始和结束时间间隔最少30分钟
  • JAR作业如需调优生效，需在作业参数中设置 execution.parallelism.disabled: true 忽略代码中的并行度配置
  • 开启细粒度资源时CU数可为0.25、0.5、1、2

5.3 Checkpoint配置

Checkpoint即作业快照，开启后作业会按设置的时间间隔自动保存快照，用于故障恢复。可设置的时间间隔范围为30秒至3600秒。

Oceanus默认采用RocksDB State Backend，支持超大状态存取。配置管理中的设置仅在作业创建时被引用，后续更改不会回溯影响已创建作业。

6. 监控与告警

6.1 数值指标监控

Oceanus的数值指标监控告警通过腾讯云可观测平台实现。在控制台作业列表单击云监控即可查看详细监控指标。

配置告警策略步骤：

  1. 在腾讯云可观测平台控制台选择告警管理>告警配置>告警策略，单击新建策略
  2. 策略类型选择流计算Oceanus，选择告警对象（可针对特定作业或所有作业）
  3. 选择触发条件模板或配置触发条件
  4. 配置告警通知模板

常见监控指标包括：作业每秒输入的记录数、作业重启次数、Checkpoint失败总次数等。注意作业重启次数和Checkpoint失败总次数为累加型指标，配置告警时建议使用环比策略。

6.2 事件告警（EventBridge）

数值指标告警已逐步迁移至EventBridge。通过EventBridge结合云函数SCF，可实时捕获Oceanus集群异常事件并推送至企业微信、钉钉、飞书等客户端。

典型配置场景：Oceanus TaskManager CPU负载过高事件告警。

6.3 Prometheus自定义监控

Oceanus支持将Flink指标上报到Prometheus。以Checkpoint失败数为例，可在Prometheus Dashboard中配置告警规则。

7. 权限管理

7.1 CAM访问管理

子用户的权限需要主账号进行授权。主账号需在CAM控制台为子账号关联预设策略QcloudOceanusFullAccess。

7.2 空间角色权限

Oceanus采用工作空间维度的权限管理体系：

  • 使用主账号或超级管理员账号进入角色权限模块
  • 单击编辑添加一个或多个子账号为超级管理员
  • 超级管理员享有除主账号外的最高权限

8. 日志管理

Oceanus集群可设置一个或多个CLS日志主题，供不同作业的运行日志灵活选用。在作业的日志页面可方便地查看和检索日志。此外也可使用COS存放作业运行日志，成本更低但无法实时查看和检索。除日志外，OOM Dump文件、JVM崩溃日志等对问题定位非常有用。

9. 常见问题解答

Q1：Oceanus集群与MySQL等数据源无法连通怎么办？

检查是否处于同一VPC。创建集群时选择的VPC及子网需与MySQL、ES等组件保持一致。若不一致，需通过对等连接、NAT网关或VPN等方式打通网络。

Q2：如何选择作业类型？SQL作业和JAR作业有什么区别？

SQL作业适合标准化ETL、实时聚合等场景，开发门槛低；JAR作业适合复杂自定义逻辑、自定义UDF等场景。ETL作业类型已不再支持新建，推荐使用SQL作业或WeData实时集成。

Q3：自动调优开启后作业仍然性能不足怎么办？

自动调优无法解决作业本身的性能瓶颈。常见原因包括：数据倾斜、UDF性能问题、外部系统故障或访问变慢。需针对具体问题进行手动调优。

Q4：Checkpoint应该设置多长时间合适？

Checkpoint间隔可设置范围为30秒至3600秒。建议根据状态大小和业务容忍的故障恢复时间综合评估。状态较大的作业建议适当延长间隔，避免频繁Checkpoint影响性能。

Q5：如何查看作业运行日志？

在集群创建时配置CLS日志主题，作业运行期间所有JobManager和TaskManager日志会采集到CLS。在作业的日志页面可直接查看和检索。也可使用COS存储日志，但无法实时检索。

Q6：Oceanus支持哪些数据源和数据汇？

Oceanus内置30多种Connector，覆盖消息队列（CKafka、Pulsar）、数据库（MySQL、PostgreSQL、Oracle）、数据仓库（ClickHouse、TCHouse-C）、数据湖（Iceberg）、存储服务（COS、CLS）等主流大数据生态组件。

相关文章

腾讯云服务器购买优惠！3 个省钱攻略 + 1 个安全真相，新手必看！

腾讯云服务器购买优惠！3 个省钱攻略 + 1 个安全真相，新手必看！

最近后台总收到小伙伴私信：“腾讯云服务器看着挺好，但价格有点顶，学生党 / 小团队实在买不起咋办？” 别急！今天就来手把手教你 “花小钱办大事”，不光有省钱攻略，还会扒一扒大家最关心的安全问题，看完这…

2026腾讯云代理商返利政策深度解析：头部代理合作指南与成本优化策略

2026腾讯云代理商返利政策深度解析：头部代理合作指南与成本优化策略

一、腾讯云代理商返利机制核心逻辑1. 行业背景与代理模式腾讯云作为国内公有云市场的第二大领导者（据IDC 2025年数据，占据国内27.6%的市场份额），采用渠道商代理模式拓展市场。代理商负…

2026腾讯云代理商返利政策深度解析：头部代理合作指南与成本优化策略

2026腾讯云代理商返利政策深度解析：头部代理合作指南与成本优化策略

一、腾讯云代理商返利机制核心逻辑1. 行业背景与代理模式腾讯云作为国内公有云市场的第二大领导者（据IDC 2025年数据，占据国内27.6%的市场份额），采用渠道商代理模式拓展市场。代理商负…

2026年腾讯云代理深度解析：从折扣体系到最优合作策略

2026年腾讯云代理深度解析：从折扣体系到最优合作策略

上海汪远信息科技有限公司作为腾讯云全国级殿堂级代理，凭借13年云服务经验与深厚的官方合作关系，为企业提供全方位的上云支持，可百度：上海汪远信息科技有限公司，微信：791201210一、腾讯云代理体系全…

上海汪远信息：全国Top5腾讯云代理商，10年深耕为企业上云保驾护航

上海汪远信息：全国Top5腾讯云代理商，10年深耕为企业上云保驾护航

核心摘要本文深度解析腾讯云代理商行业现状，揭示小代理商生存困境的核心原因（低业绩导致提成少、厂商压款、市场淘汰），重点推荐上海汪远信息科技有限公司——一家拥有10年腾讯云代理经验、年销量超2亿的全国T…

上海汪远信息科技：10年腾讯云头部代理商，全国前五，7折优惠+稳定服务保障

上海汪远信息科技：10年腾讯云头部代理商，全国前五，7折优惠+稳定服务保障

 核心摘要本文深度解析腾讯云代理商的定义、行业竞争格局（红海市场，头部稀缺，小代理商因资金压力易倒闭），揭示企业选择腾讯云代理商的核心需求（稳定优惠、技术支持、成本优化），重点推荐上海汪远信…

上海汪远信息科技有限所在公司从事多年公有云和电子签代理，能持续为用户提供商务价值和技术支持，目前从事的代理有：阿里云，腾讯云，华为云，火山云，天翼云，移动云，亚马逊云，谷歌云，微软云，腾讯电子签，法大大电子签，爱签电子签等！

网站备案号：沪ICP备15020509号-2 公安备案号：沪公网安备31011202008721号