创建数据总线DataHub源表

适用于Blink 1.4.5及以上版本。

什么是数据总线

阿里云流数据处理平台DataHub是流式数据（Streaming Data）的处理平台，提供对流式数据的发布（Publish）、订阅（Subscribe）和分发功能，让您可以轻松构建基于流式数据的分析和应用。实时计算Flink版使用DataHub作为流式数据存储源头或输出目的端。

语法示例

DataHub可以作为实时计算Flink版的数据输入，示例如下。

CREATE TABLE datahub_stream(name VARCHAR,age BIGINT,birthday BIGINT
) WITH (type='datahub',endPoint='http://dh-cn-hangzhou.aliyun-inc.com',project='<yourProjectName>',topic='<yourTopic>',accessId='<yourAccessID>',accessKey='<yourAccessSecret>',startTime='2017-07-21 00:00:00'
);

WITH参数

参数	说明	是否必填	备注
type	源表类型	是	固定值为`datahub`。
endPoint	消费端点信息	是	详情请参见DataHub域名列表。
accessId	AccessKey ID	是	无
accessKey	AccessKey Secret	是	无
project	读取的项目	是	无
topic	Project下的具体的Topic名称	是	无(注意目前只支持Tuple模式的Topic。)
startTime	启动位点的时间	否	批处理时必填，格式为`yyyy-MM-dd hh:mm:ss`。
endTime	DataHub日志结束时间	否	批处理时必填，格式为`yyyy-MM-dd hh:mm:ss`。
maxRetryTimes	读取最大重试次数	否	不同版本中，该参数的默认值如下： - Blink 2.2.7以下版本：3 - Blink 2.2.7及以上版本：20
retryIntervalMs	重试间隔	否	不同版本中，该参数的默认值如下（单位为毫秒）： - Blink 2.2.7以下版本：1000 - Blink 2.2.7及以上版本：50
batchReadSize	单次读取条数	否	默认值为10，可设置的最大值为1000。
lengthCheck	单行字段条数检查策略	否	- NONE（默认值）： 1. 解析出的字段数大于定义字段数时，按从左到右的顺序，取定义字段数量的数据。 2. 解析出的字段数小于定义字段数时，跳过这行数据。 - SKIP：解析出的字段数和定义字段数不同时跳过这行数据。 - EXCEPTION：解析出的字段数和定义字段数不同时提示异常。 - PAD：按从左到右顺序填充。 1. 解析出的字段数大于定义字段数时，按从左到右的顺序，取定义字段数量的数据。 2. 解析出的字段数小于定义字段数时，按从左到右的顺序，在行尾用Null填充缺少的字段。
columnErrorDebug	是否开启调试功能	否	- false（默认值）：关闭调试功能。 - true：开启调试功能，将打印解析异常的日志。
isBlob	DataHub是否为BLOB类型	否	默认值为false。 - 仅Blink 3.4.x 及以上版本支持该参数。 - 使用BLOB类型时，字段需要声明为VARBINARY类型，与METAQ类似。

类型映射

DataHub和实时计算Flink版字段类型对应关系如下，建议使用该对应关系时进行DDL声明。

属性字段

Flink SQL支持获取DataHub的属性字段。通过读取属性字段可以获得每条信息输入DataHub的系统时间（System Time）。

字段名	说明
timestamp	每条记录写入DataHub的系统时间（System Time）

示例代码

包含DataHub源表的实时计算Flink版作业代码示例如下。

create table datahub_input(name VARCHAR
) with (type='datahub',endPoint='http://dh-cn-hangzhou.aliyun-inc.com',project='test1',topic='topic1',accessId='<yourAccessID>',accessKey='<yourAccessSecret>',startTime='2018-06-01 00:00:00'
);create table test_out(name  VARCHAR
) with (type='print'
);INSERT INTO test_out
SELECT LOWER(name)
from datahub_input;

常见问题

Q：DataHub和实时计算Flink版中的TIMESTAMP的精确级别不同，如何进行转换？

A：DataHub的TIMESTAMP是精确到微秒，在Unix时间戳里是16位，但实时计算Flink版定义的TIMESTAMP是精确到毫秒，在Unix时间戳里是13位，所以建议您使用BIGINT映射。如果您需要使用TIMESTAMP，建议使用计算列进行转换，代码示例如下。
```
CREATE TABLE datahub_test (id                 VARCHAR,`type`             VARCHAR,`value`            DOUBLE,`time`             BIGINT,ts as TO_TIMESTAMP(`time`/1000),WATERMARK wk for ts as withoffset(ts,2000)
) WITH (type = 'datahub'
);
```
Q：分裂或者缩容DataHub Topic后导致实时计算Flink版作业失败，如何恢复？

A：Blink 2.2.0以下版本不支持DataHub Shard扩容和缩容功能，如果分裂或者缩容了某个实时计算正在读取的Topic，会导致任务持续出错，无法自行恢复。该情况下需要重新启动（停止->启动）来使任务恢复正常。
Q：可以删除正在使用的DataHub Topic吗？

A：所有版本均不支持删除或重建正在引用的DataHub Topic。

Blink SQL之创建数据总线DataHub源表相关推荐

Blink SQL之创建数据结果表
Blink创建数据结果表概述 Blink使用CREATE TABLE作为输出结果数据的格式定义,同时定义数据如何写入到目的数据结果表. 结果表有两种类型: Append类型:输出存储是日志系统.消息 ...
合肥工业大学—SQL Server数据库实验三：SQL语句创建和删除基本表
SQL语句创建和删除基本表 1. 编写6个基本表 2. 修改基本表结构,完整性约束条件 3. 用Drop table 语句删除基本表 1. 编写6个基本表设有简单教学管理的6个关系模式如下: 系(系 ...
使用SQL语句创建并管理数据表
数据库创建以后,我们需要创建数据表存储数据,表是一种重要的数据库对象.简单的叙述一下表的概念. 表的概念: 数据表是数据库中一个重要的对象,也是其他对象的基础.一个数据库中可以包含多张表,表是数据的集 ...
使用SQL语句创建数据库和数据库表
使用SQL语句创建数据库和数据库表内容如下: 创建数据库创建数据库表 1.创建数据库的语句创建数据库,需要创建数据库的数据文件和日志文件其中一些语句含义如下: 语句代表含义 name 逻辑文 ...
在MySQL中如何用SQL命令创建数据库及数据表
HRELLO 大家好! 上期给大家讲解了,MySQL数据库常用的管理命令.现在,给大家讲解,如何用SQL命令在MySQL中创建一个数据库及数据表. 在学习前,我们首先要了解,SQL命令是什么? SQL ...
Blink SQL之创建日志服务SLS源表
创建日志服务SLS源表仅适用于Blink 1.4.5及以上版本. 什么是日志服务日志服务SLS是针对日志类数据的一站式服务,对于日志服务而言,数据格式类似JSON,示例如下. {"a&q ...
Blink SQL之创建数据维表
Blink创建数据维表概述在维表DDL语法中增加1行PERIOD FOR SYSTEM_TIME的声明,定义维表的变化周期,即可使用标准的CREATE TABLE语法定义实时计算维表. 示例 CR ...
sql server2008中怎样用sql语句创建数据库和数据表
这是简单用代码实现创建数据库和数据表的sql语句,如下: --调用系统数据库-- use master go /***防止你要创建的数据库同名,先把它删除掉****/ if Exists(select ...
实验二通过SQL语句创建与管理数据表
实验目的 (1)掌握查询分析器的使用. (2)掌握通过SQL语句创建表的方法. (3)掌握通过SQL语句修改表结构的方法. (4)掌握通过SQL语句添加.修改.删除表数据的方法. 实验内容 1．通过S ...

Blink SQL之创建数据总线DataHub源表