作者：腾讯云流计算 Oceanus 团队
流计算 Oceanus 简介
流计算 Oceanus 是大数据产品生态体系的实时化分析利器，是基于 Apache Flink 构建的具备一站开发、无缝连接、亚秒延时、低廉成本、安全稳定等特点的企业级实时大数据分析平台。流计算 Oceanus 以实现企业数据价值最大化为目标，加速企业实时化数字化的建设进程。
本文将您详细介绍如何将数据写入 MongoDB。
前置准备
创建流计算 Oceanus 集群
进入 Oceanus 控制台 [1]，点击左侧【集群管理】，点击左上方【创建集群】，具体可参考 Oceanus 官方文档 创建独享集群 [2]。
创建 MongoDB 实例
进入 MongoDB 控制台 [3]，点击左上角【新建实例】创建实例，具体参考 创建 MongoDB 实例 [4]。作者这里使用 shell 的方式，下载 MongoDB 客户端的方式连接数据库，更多连接信息请参考 连接 MongoDB 实例[5]
## 安装 MongoDB 客户端wget https://fastdl.mongodb.org/linux/mongodb-linux-x86_64-rhel70-XX.XX.XX.tgz## 解压tar zxvf mongodb-linux-x86_64-rhel70-XX.XX.XX.tgz## 进入目录cd mongodb-linux-x86_64-rhel70-XX.XX.XX## 连接客户端./bin/mongo -umongouser -p***** 172.xx.xx.xx:27017/admin请注意选择与云数据库 MongoDB 服务并与 CVM 操作系统相匹配的版本
流计算 Oceanus 作业
1. 创建 Source
CREATE TABLE datagen_source_table ( id INT, name STRING) WITH ( 'connector' = 'datagen', 'rows-per-second'='1' -- 每秒产生的数据条数);2. 创建 Sink
CREATE TABLE mongodb ( id INT, name STRING) WITH ( 'connector' = 'mongodb', -- 固定值 'mongodb' 'database' = 'testdb', --数据库名 'collection' = 'test1', --数据集合 'uri' = 'mongodb://mongouser:******@xx.xx.xx.xx:27017/admin', -- MongoDB连接串 'batchSize' = '5' -- 每次批量写入的条数);3. 编写业务 SQL
INSERT INTO mongodbSELECT * from datagen_source_table;总结
本实例演示如何使用 Datagen 生成随机数据，然后使用 MongoDB Sink 连接器将数据写入 MongoDB。
目前仅 Flink 1.13 支持 Sink 端写入，其他版本暂不支持。 MongoDB Sink 暂不支持 Upsert。 MongoDB 的 User 必须拥有 database 的写权限。
参考链接
[1] Oceanus 控制台：https://console.cloud.tencent.com/oceanus/overview
[2] 创建独享集群：https://cloud.tencent.com/document/product/849/48298
[3] MongoDB 控制台：https://console.cloud.tencent.com/mongodb#/
[4] 创建 MongoDB 实例：https://cloud.tencent.com/document/product/240/3551
[5] 连接 MongoDB 实例：https://cloud.tencent.com/document/product/240/7092
扫码加入 流计算 Oceanus 产品交流群
