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

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

需要先登录腾讯云控制台，点击：

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

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

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

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

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

以下是一个使用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;

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

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

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'
);

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'
);

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'
);

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中无需提前创建类似表的实体。

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'
);

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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影响性能。

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

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

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

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

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

大数据

云计算

后端

腾讯云流计算Oceanus是基于Apache Flink构建的全托管实时计算服务，提供SQL/JAR/Python作业开发，支持30多种数据源连接器。具备亚秒级延迟、自动调优、Checkpoint容错等特性，可快速构建实时数据管道。涵盖集群创建、作业开发、性能优化全流程，适用于ETL、实时分析等场景。

日志服务

对等连接

消息队列

私有网络

对象存储

云数据库

PostgreSQL

数据库

云函数

云监控

数据湖

MySQL

对象存储COS新用户低至1元！


2026上云采购 | AI焕新·智启新局

dbdc

9.9元起，最快1天搭建自有直播平台


腾讯云数据库特惠，新用户首单0.5折起


文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

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

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

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐