云服务器内容精选
-
语法说明 ROW_NUMBER(): 从第一行开始，依次为每一行分配一个唯一且连续的号码。 PARTITION BY col1[, col2...]: 指定分区的列，例如去重的键。 ORDER BY time_attr [asc|desc]: 指定排序的列。所指定的列必须为时间属性。目前仅支持proctime。升序（ ASC ）排列指只保留第一行，而降序排列（ DESC ）则只保留最后一行。 WHERE rownum = 1: Flink 需要 rownum = 1 以确定该查询是否为去重查询。
-
示例 根据order_id对数据进行去重，其中proctime为事件时间属性列 SELECT order_id, user, product, number FROM ( SELECT *, ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY proctime ASC) as row_num FROM Orders) WHERE row_num = 1;
-
注意事项 Hive方言只能用于操作Hive对象，并要求当前Catalog是一个HiveCatalog 。 Hive方言只支持db.table这种两级的标识符，不支持带有Catalog名字的标识符。更多信息请参考Apache Flink Hive Read & Write。 虽然所有Hive版本支持相同的语法，但是一些特定的功能对Hive版本有依赖，请参考Hive 版本。 例如，更新数据库位置 只在 Hive-2.4.0 或更高版本支持。 执行DML和DQL时应该使用HiveModule 。 从Flink 1.15版本开始，在使用Hive方言抛出以下异常时，请尝试用opt目录下的 flink-table-planner_2.12 jar包来替换lib目录下的flink-table-planner-loader jar包。具体原因请参考 FLINK-25128。
-
功能描述 Avro Schema Registry (avro-confluent) 格式能让您读取被 io.confluent.kafka.serializers.KafkaAvroSerializer 序列化的记录，以及可以写入成能被 io.confluent.kafka.serializers.KafkaAvroDeserializer 反序列化的记录。 当以这种格式读取（反序列化）记录时，将根据记录中编码的 schema 版本 id 从配置的 Confluent Schema Registry 中获取 Avro writer schema ，而从 table schema 中推断出 reader schema。 当以这种格式写入（序列化）记录时，Avro schema 是从 table schema 中推断出来的，并会用来检索要与数据一起编码的 schema id。我们会在配置的 Confluent Schema Registry 中配置的 subject 下，检索 schema id。subject 通过 avro-confluent.subject 参数来指定。
-
数据类型映射 目前 Apache Flink 都是从 table schema 去推断反序列化期间的 Avro reader schema 和序列化期间的 Avro writer schema。显式地定义 Avro schema 暂不支持。 Avro Format中描述了 Flink 数据类型和 Avro 类型的对应关系。 除了此处列出的类型之外，Flink 还支持读取/写入可为空（nullable）的类型。 Flink 将可为空的类型映射到 Avro union(something, null), 其中 something 是从 Flink 类型转换的 Avro 类型。
-
参数说明 表1 参数说明 参数 是否必选 默认值 类型 说明 format 是 (none) String 指定要使用的格式，此处应为 'canal-json'. canal-json.ignore-parse-errors 否 false Boolean 当解析异常时，是跳过当前字段或行，还是抛出错误失败（默认为 false，即抛出错误失败）。如果忽略字段的解析异常，则会将该字段值设置为null。 canal-json.timestamp-format.standard 否 'SQL' String 指定输入和输出时间戳格式。当前支持的值是：'SQL'和'ISO-8601'。 选项 'SQL' 将解析 "yyyy-MM-dd HH:mm:ss.s{precision}" 格式的输入时间戳，例如 '2020-12-30 12:13:14.123'，并以相同格式输出时间戳。 选项 'ISO-8601' 将解析 "yyyy-MM-ddTHH:mm:ss.s{precision}" 格式的输入时间戳，例如 '2020-12-30T12:13:14.123'，并以相同的格式输出时间戳。 canal-json.map-null-key.mode 否 'FALL' String 指定处理 Map 中 key 值为空的方法. 当前支持的值有'FAIL', 'DROP'和 'LITERAL'。 Option 'FAIL' 将抛出异常，如果遇到 Map 中 key 值为空的数据。 Option 'DROP' 将丢弃 Map 中 key 值为空的数据项。 Option 'LITERAL' 将使用字符串常量来替换 Map 中的空 key 值。字符串常量的值由 'canal-json.map-null-key.literal' 定义。 canal-json.map-null-key.literal 否 'null' String 当 'canal-json.map-null-key.mode' 是 LITERAL 的时候，指定字符串常量替换 Map 中的空 key 值。 canal-json.encode.decimal-as-plain-number 否 false Boolean 将所有 DECIMAL 类型的数据保持原状，不使用科学计数法表示。例：0.000000027 默认会表示为 2.7E-8。当此选项设为 true 时，则会表示为 0.000000027。 canal-json.database.include 否 (none) String 一个可选的正则表达式，通过正则匹配 Canal 记录中的 "database" 元字段，仅读取指定数据库的 changelog 记录。正则字符串与 Java 的 Pattern 兼容。 canal-json.table.include 否 (none) String 一个可选的正则表达式，通过正则匹配 Canal 记录中的 "table" 元字段，仅读取指定表的 changelog 记录。正则字符串与 Java 的 Pattern 兼容。
-
功能描述 Canal是一个 CDC（ChangeLog Data Capture，变更日志数据捕获）工具，可以实时地将 MySQL 变更传输到其他系统。Canal 为变更日志提供了统一的数据格式，并支持使用 JSON 或 protobuf序列化消息（Canal 默认使用 protobuf）。 Flink 支持将 Canal 的 JSON 消息解析为 INSERT / UPDATE / DELETE 消息到 Flink SQL 系统中。在很多情况下，利用这个特性非常的有用，例如 将增量数据从数据库同步到其他系统 日志审计 数据库的实时物化视图 关联维度数据库的变更历史，等等。 Flink 还支持将 Flink SQL 中的 INSERT / UPDATE / DELETE 消息编码为 Canal 格式的 JSON 消息，输出到 Kafka 等存储中。 但需要注意的是，目前 Flink 还不支持将 UPDATE_BEFORE 和 UPDATE_AFTER 合并为一条 UPDATE 消息。因此，Flink 将 UPDATE_BEFORE 和 UPDATE_AFTER 分别编码为 DELETE 和 INSERT 类型的 Canal 消息。 更多具体使用可参考开源社区文档：Canal Format。
-
语法说明 string_split(target, separator) 表1 string_split参数说明 参数 数据类型 说明 target STRING 待处理的目标字符串。 说明： 如果target为NULL，则返回一个空行。 如果target包含两个或多个连续出现的分隔符时，则返回长度为零的空子字符串。 如果target未包含指定分隔符，则返回目标字符串。 separator VARCHAR 指定的分隔符，当前仅支持单字符分隔。
-
示例 准备测试输入数据 表2 测试源表disSource数据和分隔符 target（STRING） separator （VARCHAR） test-flink - flink - one-two-ww-three - 输入测试SQL语句 create table disSource( target STRING, separator VARCHAR ) with ( "connector.type" = "dis", "connector.region" = "xxx", "connector.channel" = "ygj-dis-in", "format.type" = 'csv' ); create table disSink( target STRING, item STRING ) with ( 'connector.type' = 'dis', 'connector.region' = 'xxx', 'connector.channel' = 'ygj-dis-out', 'format.type' = 'csv' ); insert into disSink select target, item from disSource, lateral table(string_split(target, separator)) as T(item); 查看测试结果 表3 disSink结果表数据 target（STRING） item（STRING） test-flink test test-flink flink flink flink one-two-ww-three one one-two-ww-three two one-two-ww-three ww one-two-ww-three three
-
参数说明 表1 参数 是否必选 默认值 类型 说明 format 是 (none) String 指定要使用的格式，这里应该是 'csv'。 csv.field-delimiter 否 ， String 字段分隔符 (默认',')，必须为单字符。您可以使用反斜杠字符指定一些特殊字符，例如 '\t' 代表制表符。 您也可以通过 unicode 编码在纯 SQL 文本中指定一些特殊字符，例如 'csv.field-delimiter' = '\u0001' 代表 0x01 字符。 csv.disable-quote-character 否 false Boolean 是否禁止对引用的值使用引号 (默认是 false). 如果禁止，选项 'csv.quote-character' 不能设置。 csv.quote-character 否 ‘’ String 用于围住字段值的引号字符 (默认"). csv.allow-comments 否 false Boolean 是否允许忽略注释行（默认不允许），注释行以 '#' 作为起始字符。 如果允许注释行，请确保 csv.ignore-parse-errors 也开启了从而允许空行。 csv.ignore-parse-errors 否 false Boolean 当解析异常时，是跳过当前字段或行，还是抛出错误失败（默认为 false，即抛出错误失败）。如果忽略字段的解析异常，则会将该字段值设置为null。 csv.array-element-delimiter 否 ； String 分隔数组和行元素的字符串(默认';'). csv.escape-character 否 (none) String 转义字符(默认关闭). csv.null-literal 否 (none) String 是否将 "null" 字符串转化为 null 值。
-
常见问题 Q：Flink作业运行失败，作业运行日志中如下报错信息，应该怎么解决？ org.apache.zookeeper.ClientCnxn$SessionTimeoutException: Client session timed out, have not heard from server in 90069ms for connection id 0x0 A：可能是跨源连接未绑定或跨源绑定失败。参考增强型跨源连接重新配置跨源，Kafka集群安全组放通 DLI 队列的网段地址。
-
注意事项 创建Flink OpenSource SQL作业时，在作业编辑界面的“运行参数”处，“Flink版本”需要选择“1.12”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。 创建的HBase结果表的列簇必须定义为ROW类型，字段名对应列簇名（column family），嵌套的字段名对应列限定符名（column qualifier）。用户只需在表结构中声明查询中使用的的列簇和列限定符。除了ROW类型的列，剩下的原子数据类型字段（比如，STRING, BIGINT）将被识别为 HBase的rowkey，一张表中只能声明一个rowkey。rowkey字段的名字可以是任意的，如果是保留关键字，需要用反引号。
-
语法格式 create table hbaseSink ( attr_name attr_type (',' attr_name attr_type)* ','PRIMARY KEY (attr_name, ...) NOT ENFORCED) ) with ( 'connector' = 'hbase-2.2', 'table-name' = '', 'zookeeper.quorum' = '' );
-
功能描述 DLI将作业的输出数据输出到HBase中。HBase是一个稳定可靠，性能卓越、可伸缩、面向列的分布式 云存储 系统，适用于海量数据存储以及分布式计算的场景，用户可以利用HBase搭建起TB至PB级数据规模的存储系统，对数据轻松进行过滤分析，毫秒级得到响应，快速发现数据价值。HBase支持消息数据、报表数据、推荐类数据、风控类数据、日志数据、订单数据等结构化、半结构化的KeyValue数据存储。 利用DLI，用户可方便地将海量数据高速、低时延写入HBase。
更多精彩内容
CDN加速
GaussDB
文字转换成语音
免费的服务器
如何创建网站
域名网站购买
私有云桌面
云主机哪个好
域名怎么备案
手机云电脑
SSL证书申请
云点播服务器
免费OCR是什么
电脑云桌面
域名备案怎么弄
语音转文字
文字图片识别
云桌面是什么
网址安全检测
网站建设搭建
国外CDN加速
SSL免费证书申请
短信批量发送
图片OCR识别
云数据库MySQL
个人域名购买
录音转文字
扫描图片识别文字
OCR图片识别
行驶证识别
虚拟电话号码
电话呼叫中心软件
怎么制作一个网站
Email注册网站
华为VNC
图像文字识别
企业网站制作
个人网站搭建
华为云计算
免费租用云托管
云桌面云服务器
ocr文字识别免费版
HTTPS证书申请
图片文字识别转换
国外域名注册商
使用免费虚拟主机
云电脑主机多少钱
鲲鹏云手机
短信验证码平台
OCR图片文字识别
SSL证书是什么
申请企业邮箱步骤
免费的企业用邮箱
云免流搭建教程
域名价格
