实时计算Flink > 独享模式 > Batch(试用) > 创建源表 —— 创建DataHub源表 请输入关键词

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: 本页目录 创建DataHub源表 数据总线(DATAHUB) 属性字段 WITH参数 类型映射 创建DataHub源表 数据总线(DATAHUB) DataHub作为一个流式数据总线,为阿里云数加平台提供了大数据的入口服务。

创建DataHub源表

数据总线(DATAHUB)

DataHub作为一个流式数据总线,为阿里云数加平台提供了大数据的入口服务。结合阿里云众多云产品,可以构建一站式的数据处理平台。流计算通常使用DataHub作为流式数据存储头和输出目的端。同时,上游众多流式数据,包括DTS、IOT等均选择DataHub作为大数据平台的数据入口。

DataHub本身是流数据存储,Flink可将其作为批处理的输入。示例如下:

 
  
  1. create table datahub_stream(
  2. name varchar,
  3. age BIGINT,
  4. birthday BIGINT
  5. ) with (
  6. type='datahub',
  7. endPoint='http://dh-et2.aliyun-inc.com',
  8. project='blink_datahub_test',
  9. topic='test_topic_1',
  10. accessId='0i70RRFJXXXX',
  11. accessKey='yF60EwURseo1UAn4NiXXXXXhCfHU',
  12. startTime='2017-07-21 00:00:00',
  13. endTime='2017-07-21 01:00:00',
  14. blinkEnvironmentTypeKey='batchExec'
  15. );

注意: 作为批处理的输入时,必须指定起始及结束时间。

属性字段

目前默认支持的属性字段如下,也支持其他自定义写入的字段:

字段名 注释说明
timestamp 每条记录入datahub的systemtime

属性字段的定义和获取详见获取源表属性字段

WITH参数

目前只支持tuple模式的topic。

参数 注释说明 备注
endPoint 消费端点信息
accessId 读取的accessId
accessKey 读取的密钥
project 读取的项目
topic project下的具体的topic
startTime 日志开始时间 格式为yyyy-MM-dd hh:mm:ss
maxRetryTimes 读取最大尝试次数 可选,默认为20。
retryIntervalMs 重试间隔 可选,默认为1000。
batchReadSize 单次读取条数 可选,默认为10。
lengthCheck 单行字段条数检查策略 可选,默认为SKIP。其它可选值为EXCEPTIONPADSKIP表示字段数目不符合时跳过 。 EXCEPTION表示字段数目不符合时抛出异常。 PAD表示按顺序填充,不存在的置为null。
columnErrorDebug 是否打开调试开关,如果打开,会把解析异常的log打印出来。 可选,默认为false
isBlob datahub是否为blob类型 可选,默认为false。2.0.x版本开始支持
endTime datahub日志数据结束时间 批处理方式读取datahub数据时,必须声明该参数。

注意:使用blob类型时,字段需要声明成varbinary类型,与metaq类似。

类型映射

DataHub和流计算字段类型对应关系,建议您使用该对应关系进行DDL声明。

DataHub字段类型 流计算字段类型
BIGINT BIGINT
DOUBLE DOUBLE
TIMESTAMP BIGINT
BOOLEAN BOOLEAN
DECIMAL DECIMAL
本文转自实时计算—— 创建DataHub源表
相关实践学习
实时数据及离线数据上云方案
本实验通过使用CANAL、DataHub、DataWorks、MaxCompute服务,实现数据上云,解决了数据孤岛问题,同时把数据迁移到云计算平台,对后续数据的计算和应用提供了第一步开山之路。
相关文章
|
2月前
|
定位技术 流计算
在Flink CEP中,可以通过定义带有时间约束的模式来匹配事件的持续时间
在Flink CEP中,可以通过定义带有时间约束的模式来匹配事件的持续时间
24 2
|
1月前
|
API 数据库 流计算
有大佬知道在使用flink cdc实现数据同步,如何实现如果服务停止了对数据源表的某个数据进行删除操作,重启服务之后目标表能进行对源表删除的数据进行删除吗?
【2月更文挑战第27天】有大佬知道在使用flink cdc实现数据同步,如何实现如果服务停止了对数据源表的某个数据进行删除操作,重启服务之后目标表能进行对源表删除的数据进行删除吗?
52 3
|
4月前
|
Kubernetes Java 流计算
Flink application on k8s 有没有和 session 模式通过-C 指定额外的 jar 的相同功能啊?
Flink application on k8s 有没有和 session 模式通过-C 指定额外的 jar 的相同功能啊?
30 0
|
1月前
|
定位技术 流计算
在Flink CEP中,可以通过定义带有时间约束的模式来匹配事件的持续时间
【2月更文挑战第12天】在Flink CEP中,可以通过定义带有时间约束的模式来匹配事件的持续时间
34 3
|
2月前
|
Java 流计算
在Flink实时任务中,POJO(Plain Old Java Object)对象的模式演进可能会引起不兼容的问题
【2月更文挑战第6天】在Flink实时任务中,POJO(Plain Old Java Object)对象的模式演进可能会引起不兼容的问题
19 3
|
2月前
|
消息中间件 SQL Java
flink问题之Application 模式下启动失败如何解决
Apache Flink是由Apache软件基金会开发的开源流处理框架,其核心是用Java和Scala编写的分布式流数据流引擎。本合集提供有关Apache Flink相关技术、使用技巧和最佳实践的资源。
49 0
|
2月前
|
资源调度 Kubernetes Java
Flink--day02、Flink部署(Yarn集群搭建下的会话模式部署、单作业模式部署、应用模式部署)
Flink--day022、Flink部署(Yarn集群搭建下的会话模式部署、单作业模式部署、应用模式部署)
134 5
|
2月前
|
消息中间件 SQL 关系型数据库
Flink数据源问题合集之设置成批量模式如何解决
Flink数据源是指Apache Flink用于读取外部系统数据的接口或组件;本合集将探讨Flink数据源的类型、配置方法和最佳实践,以及在使用数据源时可能遇到的错误和解决方案。
31 2
|
3月前
|
存储 监控 Apache
Flink整库同步 如何把数据丢入到 datahub 中
Flink整库同步 如何把数据丢入到 datahub 中
|
3月前
|
资源调度 监控 API
在Flink中,通过YARN模式进行峰谷动态并行度扩容缩容可以使用 Flink 自带的动态调优功能
在Flink中,通过YARN模式进行峰谷动态并行度扩容缩容可以使用 Flink 自带的动态调优功能【1月更文挑战第6天】【1月更文挑战第26篇】
146 1