华为云用户手册

异构数据源联邦分析 车企数字化服务转型 面临市场新的竞争压力及出行服务不断变革，车企通过构建车联云平台和车机OS，将互联网应用与用车场景打通，完成车企数字化服务转型，从而为车主提供更好的智联出行体验，增加车企竞争力，促进销量增长。例如：通过对车辆日常指标数据（电池、发动机，轮胎胎压、安全气囊等健康状态）的采集和分析，及时将维保建议回馈给车主。 优势 多源数据分析免搬迁：关系型数据库RDS中存放车辆和车主基本信息， 表格存储 CloudTable中存放实时的车辆位置和健康状态信息， 数据仓库 DWS中存放周期性统计的指标。通过 DLI 无需数据搬迁，对多数据源进行联邦分析。 数据分级存储：车企需要保留全量历史数据支撑审计类等业务，低频进行访问。温冷数据存放在低成本的 对象存储服务 OBS上，高频访问的热数据存放在数据引擎（CloudTable和DWS）中，降低整体存储成本。 告警快速敏捷触发服务器弹性伸缩：对CPU、内存、硬盘空间和带宽无特殊要求。 建议搭配以下服务使用 DIS、 CDM 、OBS、DWS、RDS、CloudTable 图2 车企数字化服务转型

大数据ETL处理 运营商大数据分析 运营商数据体量在PB~EB级，其数据种类多，有结构化的基站信息数据，非结构化的消息通信数据，同时对数据的时效性有很高的要求，DLI服务提供批处理、流处理等多模引擎，打破数据孤岛进行统一的数据分析。 优势 大数据ETL：具备TB~EB级运营商 数据治理 能力，能快速将海量运营商数据做ETL处理，为分布式批处理计算提供分布式数据集。 高吞吐低时延：采用Apache Flink的Dataflow模型，高性能计算资源，从用户自建的Kafka、 MRS -Kafka、DMS-Kafka消费数据，单CU每秒吞吐1千~2万条消息。 细粒度权限管理：P公司内部有N个子部门，子部门之间需要对数据进行共享和隔离。DLI支持计算资源按租户隔离，保障作业SLA；支持数据权限控制到表/列，帮助企业实现部门间数据共享和权限管理。 建议搭配以下服务使用 OBS、DIS、 DataArts Studio 图3 运营商大数据分析

地理大数据分析 地理大数据分析 地理大数据具有大数据的相关特征，数据体量巨大，例如，全球卫星遥感影像数据量达到PB级。数据种类多，有结构化的遥感影像栅格数据、矢量数据，非结构化的空间位置数据、三维建模数据；在大体量的地理大数据中，通过高效的挖掘工具或者挖掘方法实现价值提炼，是用户非常关注的话题。 优势 提供地理专业算子：支持全栈Spark能力，具备丰富的Spark空间数据分析算法算子，全面支持结构化的遥感影像数据、非结构化的三维建模、激光点云等巨量数据的离线批处理，支持带有位置属性的动态流数据实时计算处理。 CEP SQL：提供地理位置分析函数对地理空间数据进行实时分析，用户仅需编写SQL便可实现例如偏航检测，电子围栏等地理分析场景。 大数据治理能力：能快速将海量遥感影像数据接入上云，快速完成影像数据切片处理，为分布式批处理计算提供弹性分布式数据集。 建议搭配以下服务使用 DIS、CDM、DES、OBS、RDS、CloudTable 图4 地理大数据分析

海量 日志分析 游戏运营数据分析 游戏公司不同部门日常通过游戏数据分析平台，分析每日新增日志获取所需指标，通过数据来辅助决策。例如：运营部门通过平台获取新增玩家、活跃玩家、留存率、流失率、付费率等，了解游戏当前状态及后续响应活动措施；投放部门通过平台获取新增玩家、活跃玩家的渠道来源，来决定下一周期重点投放哪些平台。 优势 高效的Spark编程模型：使用DLI直接从DIS中获取数据，进行数据清理等预处理操作。只需编写处理逻辑，无需关心多线程模型。 简单易用：直接使用标准SQL编写指标分析逻辑，无需关注背后复杂的分布式计算平台。 按需计费：日志分析按时效性要求按周期进行调度，每次调度之间存在大量空闲期。DLI按需计费只在使用期间收费，成本较独占队列降低50%以上。 建议搭配以下服务使用 OBS，DIS，DWS，RDS 图1 游戏运营数据分析

扫描量计费模式 扫描量支持以下计费模式： 按需计费：在default队列中提交作业按扫描数据量计费，一种按需计费模式。 计费方式自创建起按自然小时收费。按小时结算，小时数按整点计算。 扫描量计费=单价*扫描量数据（GB）。 套餐包： 数据湖探索 支持扫描数据量套餐包，购买了扫描数据量套餐包，按需使用过程中优先抵扣扫描数据量套餐包的规格额度，超过扫描数据量套餐包额度的按照按需计费。扫描数据量套餐的额度每个月会重置。 更多套餐包计费信息请参考套餐包计费。 本文将介绍按需计费的DLI数据扫描量计费的计费规则。

约束限制 仅default队列适用按数据扫描量计费。 如果扫描字节数量少于10M， 则按10M计算。 数据定义语言（DDL）语句（如 CREATE、ALTER、DROP TABLE等语句）、管理分区语句、失败的查询语句无需付费。 取消的查询，按取消查询时扫描的总数据量计费。 若计算任务超时或失败，则当次计算不收取费用。 数据的存储方式（包括格式，是否压缩）对计算费用有直接的影响。 假设Parquet格式文件为原 CS V文件大小的1/4，则数据扫描费用为原CSV文件扫描费用的1/4。

续费相关的功能 包年/包月的DLI资源续费相关的功能如表1所示。 表1 续费相关的功能 功能 说明 手动续费 包年/包月的DLI资源从购买到被自动删除之前，您可以随时在DLI控制台为队列或弹性资源池续费，以延长资源的使用时间。 自动续费 开通自动续费后，DLI资源会在每次到期前自动续费，避免因忘记手动续费而导致资源被自动删除。 在DLI资源生命周期的不同阶段，您可以根据需要选择一种方式进行续费，具体如图1所示。 图1 DLI资源生命周期 DLI资源从购买到到期前，处于正常运行阶段，资源状态为“运行中”。 到期后，资源状态变为“已过期”。 到期未续费时，DLI资源首先会进入宽限期，宽限期到期后仍未续费，资源状态变为“已冻结”。 超过宽限期仍未续费将进入保留期，如果保留期内仍未续费，资源将被自动删除。 华为云根据客户等级定义了不同客户的宽限期和保留期时长。 在DLI资源到期前均可开通自动续费，到期前7日凌晨3:00首次尝试自动续费，如果扣款失败，每天凌晨3:00尝试一次，直至DLI资源到期或者续费成功。到期前7日自动续费扣款是系统默认配置，您也可以根据需要修改此扣款日。

续费简介 包年/包月DLI资源到期后会影响资源的正常使用。如果您想继续使用，需要在指定的时间内续费，否则队列、弹性资源池等资源会自动释放，数据丢失且不可恢复。 续费操作仅适用于包年/包月资源，按需计费资源不需要续费，只需要保证账户余额充足即可。DLI的包周期资源包括：队列、弹性资源池。 在到期前续费成功，所有资源得以保留，且作业的运行不受影响。资源到期后的状态说明，请参见到期后影响（适用于包年/包月计费模式）。

欠费影响 包年/包月 对于包年/包月DLI资源。如队列或弹性资源池，用户已经预先支付了资源费用，因此在账户出现欠费的情况下，已有的包年/包月资源仍可正常使用。然而，对于涉及费用的操作，如新购资源、弹性资源池规格变更、续费订单等，用户将无法正常进行。 按需计费 当您的账号因按需DLI资源自动扣费导致欠费后，账号将变成欠费状态。欠费后，资源进入宽限期，不能在DLI服务中提交作业，包括SQL作业，Spark作业和Flink作业。 如果您在宽限期内仍未支付欠款，那么就会进入保留期，资源状态变为“已冻结”，您将无法对处于保留期的按需计费资源执行任何操作。 保留期到期后，若您仍未支付账户欠款，那么DLI的按需资源都将被释放，数据无法恢复。 图1 按需计费DLI资源生命周期 华为云根据客户等级定义了不同客户的宽限期和保留期时长。

DLI的计费模式有哪些？ DLI在不同的计费项下有不同的计费模式，具体详见表1说明。 表1 DLI计费模式 计费项 计费模式 计费模式说明 计算计费 包年/包月 按照弹性资源池包年/包月的固定费用计费。 按需计费 按需弹性资源池资源使用CU时进行计费。 计算费用=单价*CU数*小时数 CU时套餐包 DLI提供了弹性资源池CU时套餐包。 CU时套餐包的额度会按订购周期重置。 DLI表的数据存储 按需计费 按照存储在DLI服务中的数据存储量（单位为“GB”）收取存储费用。 存储费用=单价*存储数据量（GB）*小时数 存储套餐包 购买了存储量套餐包，按需使用过程中优先抵扣存储套餐包的规格额度，超过套餐包额度的按照按需计费。存储套餐的额度每个小时会重置。 数据扫描量 按需计费 扫描量计费=单价*扫描量数据（GB） 扫描数据量套餐包 购买了扫描数据量套餐包，按需使用过程中优先抵扣扫描数据量套餐包的规格额度，超过扫描数据量套餐包额度的按照按需计费。扫描数据量套餐的额度会按订购周期重置。 创建队列的操作指导请参考：创建队列 具体的计费模式请参考：《计费说明》。 具体价格请参考：《 数据湖 探索价格详情》。 父主题： 计费相关问题

包年/包月资源 对于包年/包月计费模式的资源，例如包年/包月的弹性资源池、包年/包月的队列等，用户在购买时会一次性付费，服务将在到期后自动停止使用。 如果在计费周期内不再使用包年/包月资源，您可以执行退订操作，系统将根据资源是否属于五天无理由退订、是否使用代金券和折扣券等条件返还一定金额到您的账户。详细的退订规则请参见云服务退订规则概览。 如果您已开启“自动续费”功能，为避免继续产生费用，请在自动续费扣款日（默认为到期前7日）之前关闭自动续费。

按需计费资源 对于按需计费模式的资源： 按需计费的弹性资源，若不再使用这些资源且需停止计费，请删除相应资源。 按需计费的数据存储仅在存储DLI表时计费，若不再使用这些资源且需停止计费，请删除相应资源。 按需计费的扫描量仅在使用default队列执行作业时候按扫描量计费，作业执行完成后停止计费。 按需计费资源删除后，可能还会存在账单信息，因为系统通常会在使用后1个小时内对上一个结算周期的费用进行扣款。例如在8:30删除按小时结算的资源，但是8:00~9:00期间产生的费用，通常会在10:00左右才进行扣费。

步骤8：查看作业运行结果 在Spark作业管理界面显示已提交的作业运行状态。初始状态显示为“启动中”。 如果作业运行成功则作业状态显示为“已成功”，单击“操作”列“更多”下的“Driver日志”，显示当前作业运行的日志。 图16 diver日志 图17 “Driver日志”中的作业执行日志 如果作业运行成功，本示例进入OBS桶下的“result/parquet”目录，查看已生成预期的parquet文件。 图18 obs桶文件 如果作业运行失败，单击“操作”列“更多”下的“Driver日志”，显示具体的报错日志信息，根据报错信息定位问题原因。 例如，如下截图信息因为创建Spark Jar作业时主类名没有包含包路径，报找不到类名“SparkDemoObs”。 图19 报错信息 可以在“操作”列，单击“编辑”，修改“主类”参数为正确的：com.huawei.dli.demo.SparkDemoObs，单击“执行”重新运行该作业即可。

语法格式 create table tableName( attr_name attr_type (',' attr_name attr_type)* (',' WATERMARK FOR rowtime_column_name AS watermark-strategy_expression) ) with ( ... 'dew.endpoint'='', 'dew.csms.secretName'='', 'dew.csms.decrypt.fields'='', 'dew.projectId'='', 'dew.csms.version'='' );

参数说明 表1 参数说明 参数 是否必选 默认值 数据类型 参数说明 dew.endpoint 是 无 String 指定要使用的DEW服务所在的endpoint信息。 获取地区和终端节点。 配置示例：'dew.endpoint'='kms.cn-xxxx.myhuaweicloud.com' dew.projectId 否 有 String DEW所在的项目ID， 默认是Flink作业所在的项目ID。 获取项目ID dew.csms.secretName 是 无 String 在DEW服务的凭据管理中新建的通用凭据的名称。 配置示例：'dew.csms.secretName'='secretInfo' dew.csms.decrypt.fields 是 无 String 指定connector with属性中，哪些字段属性需要使用DEW云凭据管理服务进行解密。字段属性之间用逗号分隔，例如：'dew.csms.decrypt.fields'='field1,field2,field3' dew.csms.version 否 最新的version String 在DEW服务的凭据管理中新建的通用凭据的版本号(凭据的版本标识符)。 若不指定，则默认获取该通用凭证的最新版本号。 配置示例： 'dew.csms.version'='v1'

示例 本例以通过DataGen表产生随机数据并输出到Mysql结果表中为例，介绍Flink Opensource SQL使用DEW管理访问凭据的配置方法。 创建DLI与Mysql的增强型跨源连接。详细步骤请参考创建增强型跨源连接。 创建DLI访问DEW的委托并完成委托授权。详细步骤请参考自定义DLI委托权限。 在DEW创建通用凭证。详细操作请参考创建通用凭据。 登录DEW管理控制台 选择“凭据管理”，进入“凭据管理”页面。 单击“创建凭据”。配置凭据基本信息 本例配置Mysql的凭据值： "MySQLUsername":"demo" "MySQLPassword":"*******",其中"*******"为Mysql的访问密码 Flink作业编辑框中输入具体的作业SQL，本示例作业参考如下。 create table dataGenSource( user_id string, amount int ) with ( 'connector' = 'datagen', 'rows-per-second' = '1', --每秒生成一条数据 'fields.user_id.kind' = 'random', --为字段user_id指定random生成器 'fields.user_id.length' = '3' --限制user_id长度为3 ); CREATE TABLE jdbcSink ( user_id string, amount int ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://MySQLAddress:MySQLPort/flink',--其中url中的flink表示MySQL中orders表所在的数据库名 'table-name' = 'orders', 'username' = 'MySQLUsername', -- DEW服务中，名称为secretInfo，且版本号v1的的通用凭证，定义凭证值的键MySQLUsername，它的值为用户的敏感信息。 'password' = 'MySQLPassword', -- DEW服务中，名称为secretInfo，且版本号v1的的通用凭证，定义凭证值的键MySQLPassword，它的值为用户的敏感信息。 'sink.buffer-flush.max-rows' = '1', 'dew.endpoint'='kms.cn-xxxx.myhuaweicloud.com', --使用的DEW服务所在的endpoint信息 'dew.csms.secretName'='secretInfo', --DEW服务通用凭据的凭据名称 'dew.csms.decrypt.fields'='username,password', --其中username,password字段值，需要利用DEW凭证管理,进行解密替换。 'dew.csms.version'='v1' ); insert into jdbcSink select * from dataGenSOurce;

前提条件 已在DEW服务创建通用凭证，并存入凭据值。具体操作请参考：创建通用凭据。 已创建DLI访问DEW的委托并完成委托授权。该委托需具备以下权限： DEW中的查询凭据的版本与凭据值ShowSecretVersion接口权限，csms:secretVersion:get。 DEW中的查询凭据的版本列表ListSecretVersions接口权限，csms:secretVersion:list。 DEW解密凭据的权限，kms:dek:decrypt。 委托权限示例请参考自定义DLI委托权限和常见场景的委托权限策略。 仅支持Flink1.15版本使用DEW管理访问凭据，在创建作业时，请配置作业使用Flink1.15版本、且已在作业中配置允许DLI访问DEW的委托信息。 在DLI管理控制台新建“增强型跨源连接”配置DLI与数据源的网络连通。 具体操作请参考增强型跨源连接。

操作场景 DLI将Flink作业的输出数据写入到Mysql或DWS时，需要在connector中设置账号、密码等属性。但是账号密码等信息属于高度敏感数据，需要做加密处理，以保障用户的数据隐私安全。 数据加密 服务（Data Encryption Workshop，DEW）、云凭据管理服务（Cloud Secret Management Service，C SMS ），提供一种安全、可靠、简单易用隐私数据加解密方案。 用户或应用程序通过凭据管理服务，创建、检索、更新、删除凭据，轻松实现对敏感凭据的全生命周期的统一管理，有效避免程序硬编码或明文配置等问题导致的敏感信息泄露以及权限失控带来的业务风险。 本节操作介绍Flink Opensource SQL场景使用DEW管理和访问凭据的操作指导。

功能描述 窗口 Top-N 是特殊的 Top-N，它返回每个分区键的每个窗口的N个最小或最大值。 与普通Top-N不同，窗口Top-N只在窗口最后返回汇总的Top-N数据，不会产生中间结果。窗口 Top-N 会在窗口结束后清除不需要的中间状态。 窗口 Top-N 适用于用户不需要每条数据都更新Top-N结果的场景，相对普通Top-N来说性能更好。通常，窗口 Top-N 直接用于窗口表值函数（Windowing TVFs）窗口 Top-N 可以用于基于窗口表值函数（Windowing TVFs）的操作之上，比如窗口聚合，窗口Top-N和 窗口关联。 窗口 Top-N 的语法和普通的 Top-N 相同。 除此之外，窗口 Top-N 需要 PARTITION BY 子句包含窗口表值函数或窗口聚合产生的 window_start 和 window_end。 否则优化器无法翻译。 更多介绍和使用请参考开源社区文档：窗口Top-N。

常见问题 Q：MySQL CDC源表不支持定义Watermark，怎么进行窗口聚合？ A：可以采用非窗口聚合的方式，即将时间字段转换成窗口值，然后根据窗口值进行GROUP BY聚合。 例如：基于上述示例，统计每分钟的订单数，脚本如下（其中order_time为string类型，表示订单的时间）。 insert into printSink select DATE_FORMAT(order_time, 'yyyy-MM-dd HH:mm'), count(*) from mysqlCdcSource group by DATE_FORMAT(order_time, 'yyyy-MM-dd HH:mm');

元数据 元数据可以在 DDL 中作为只读（虚拟）meta 列声明。 表3 元数据 Key 数据类型 说明 table_name STRING NOT NULL 当前记录所属的表名称。 database_name STRING NOT NULL 当前记录所属的库名称。 op_ts TIMESTAMP_LTZ(3) NOT NULL 当前记录表在数据库中更新的时间。 如果从表的快照而不是 binlog 读取记录，该值将始终为0。

语法格式 create table mySqlCdcSource ( attr_name attr_type (',' attr_name attr_type)* (','PRIMARY KEY (attr_name, ...) NOT ENFORCED) ) with ( 'connector' = 'mysql-cdc', 'hostname' = 'mysqlHostname', 'username' = 'mysqlUsername', 'password' = 'mysqlPassword', 'database-name' = 'mysqlDatabaseName', 'table-name' = 'mysqlTableName' );

注意事项 创建Flink OpenSource SQL作业时，在作业编辑界面的“运行参数”处，“Flink版本”需要选择“1.15”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。 认证用的username和password等硬编码到代码中或者明文存储都有很大的安全风险，建议使用DEW管理凭证。配置文件或者环境变量中密文存放，使用时解密，确保安全。Flink Opensource SQL使用DEW管理访问凭据 为每个 Reader 设置不同的 Server id 每个用于读取 binlog 的 MySQL 数据库客户端都应该有一个唯一的 id，称为 Server id。 MySQL 服务器将使用此 id 来维护网络连接和 binlog 位置。 因此，如果不同的作业共享相同的 Server id， 则可能导致从错误的 binlog 位置读取数据。 因此，建议通过为每个 Reader 设置不同的 Server id SQL Hints, 假设 Source 并行度为 4, 我们可以使用 SELECT * FROM source_table /*+ OPTIONS('server-id'='5401-5404') */ ; 来为 4 个 Source readers 中的每一个分配唯一的 Server id。 设置 MySQL 会话超时时间 当为大型数据库创建初始一致快照时，你建立的连接可能会在读取表时碰到超时问题。你可以通过在 MySQL 侧配置 interactive_timeout 和 wait_timeout 来缓解此类问题。 interactive_timeout: 服务器在关闭交互连接之前等待活动的秒数。 更多信息请参考 MySQL documentations. wait_timeout: 服务器在关闭非交互连接之前等待活动的秒数。 更多信息请参考 MySQL documentations. 使用无主键表必须设置 scan.incremental.snapshot.chunk.key-column，且只能选择非空类型的一个字段。 在使用无主键表时，需要注意以下两种情况： 配置 scan.incremental.snapshot.chunk.key-column 时，如果表中存在索引，请尽量使用索引中的列来加快 select 度。 无主键表的处理语义由 scan.incremental.snapshot.chunk.key-column 指定的列的行为决定： 如果指定的列不存在更新操作，此时可以保证 Exactly once 语义。 如果指定的列存在更新操作，此时只能保证 At least once 语义。但可以结合下游，通过指定下游主键，结合幂等性操作来保证数据的正确性。 MySQL CDC源表暂不支持定义Watermark。如果您需要进行窗口聚合，请参考常见问题描述。 若连接DWS、MySQL等支持upsert的sink源，需要在sink表的创建语句中定义主键，请参考示例中printSink建表语句。

参数说明 表2 源表参数说明 参数 是否必选 默认值 数据类型 说明 connector 是 无 String connector类型，需配置为'mysql-cdc'。 hostname 是 无 String MySQL 数据库服务器的 IP 地址或主机名。 username 是 无 String 连接到 MySQL 数据库服务器时要使用的 MySQL 用户的名称。 password 是 无 String 连接 MySQL 数据库服务器时使用的密码。 database-name 是 无 String 要监视的 MySQL 服务器的数据库名称。 数据库名称还支持正则表达式，以监视多个与正则表达式匹配的表。 前缀匹配：^(test).* 匹配前缀为test的数据库名，例如test1、test2等。 后缀匹配：.*[p$] 匹配后缀为p的数据库名，例如cdcp、edcp等。 特定匹配：txc 匹配具体的数据库名。 table-name 是 无 String 需要监视的 MySQL 数据库的表名。表名支持正则表达式，以监视满足正则表达式的多个表。 注意：MySQL CDC 连接器在正则匹配表名时，会把用户填写的 database-name， table-name 通过字符串 `\\.` 连接成一个全路径的正则表达式，然后使用该正则表达式和 MySQL 数据库中表的全限定名进行正则匹配。 前缀匹配：^(test).* 匹配前缀为test的表名，例如test1、test2等。 后缀匹配：.*[p$] 匹配后缀为p的表名，例如cdcp、edcp等。 特定匹配：txc 匹配具体的表名。 port 否 3306 Integer MySQL 数据库服务器的整数端口号。 server-id 否 无 String 读取数据使用的 server id，server id 可以是个整数或者一个整数范围，比如 '5400' 或 '5400-5408'。 建议在 'scan.incremental.snapshot.enabled' 参数为启用时，配置成整数范围。因为在当前 MySQL 集群中运行的所有 slave 节点，标记每个 salve 节点的 id 都必须是唯一的。 所以当连接器加入 MySQL 集群作为另一个 slave 节点（并且具有唯一 id 的情况下），它就可以读取 binlog。 默认情况下，连接器会在 5400 和 6400 之间生成一个随机数，但是我们建议用户明确指定 Server id。 scan.incremental.snapshot.enabled 否 true Boolean 增量快照是一种读取表快照的新机制，与旧的快照机制相比， 增量快照有许多优点，包括： 在快照读取期间，Source 支持并发读取 在快照读取期间，Source 支持进行 chunk 粒度的 checkpoint 在快照读取之前，Source 不需要数据库锁权限。 如果希望 Source 并行运行，则每个并行 Readers 都应该具有唯一的 Server id，所以 Server id 必须是类似 `5400-6400` 的范围，并且该范围必须大于并行度。 scan.incremental.snapshot.chunk.size 否 8096 Integer 表快照的块大小（行数），读取表的快照时，捕获的表被拆分为多个块。 scan.snapshot.fetch.size 否 1024 Integer 读取表快照时每次读取数据的最大条数。 scan.startup.mode 否 initial String MySQL CDC 消费者可选的启动模式， 合法的模式为 "initial"，"earliest-offset"，"latest-offset"，"specific-offset" 和 "timestamp"。 initial （默认）：在第一次启动时对受监视的数据库表执行初始快照，并继续读取最新的 binlog。 earliest-offset：跳过快照阶段，从可读取的最早 binlog 位点开始读取。 latest-offset：首次启动时，从不对受监视的数据库表执行快照， 连接器仅从 binlog 的结尾处开始读取，这意味着连接器只能读取在连接器启动之后的数据更改。 specific-offset：跳过快照阶段，从指定的 binlog 位点开始读取。位点可通过 binlog 文件名和位置指定，或者在 GTID 在集群上启用时通过 GTID 集合指定。 timestamp：跳过快照阶段，从指定的时间戳开始读取 binlog 事件。 scan.startup.specific-offset.file 否 无 String 在 "specific-offset" 启动模式下，启动位点的 binlog 文件名。 scan.startup.specific-offset.pos 否 无 Long 在 "specific-offset" 启动模式下，启动位点的 binlog 文件位置。 scan.startup.specific-offset.gtid-set 否 无 String 在 "specific-offset" 启动模式下，启动位点的 GTID 集合。 scan.startup.specific-offset.skip-events 否 无 Long 在指定的启动位点后需要跳过的事件数量。 scan.startup.specific-offset.skip-rows 否 无 Long 在指定的启动位点后需要跳过的数据行数量。 server-time-zone 否 无 String 数据库服务器中的会话时区， 例如： "Asia/Shanghai". 它控制 MYSQL 中的时间戳类型如何转换为字符串。 更多请参考 这里. 如果没有设置，则使用ZoneId.systemDefault()来确定服务器时区。 debezium.min.row. count.to.stream.result 否 1000 Integer 在快照操作期间，连接器将查询每个包含的表，以生成该表中所有行的读取事件。 此参数确定 MySQL 连接是否将表的所有结果拉入内存（速度很快，但需要大量内存）， 或者结果是否需要流式传输（传输速度可能较慢，但适用于非常大的表）。 该值指定了在连接器对结果进行流式处理之前，表必须包含的最小行数，默认值为1000。 将此参数设置为`0`以跳过所有表大小检查，并始终在快照期间对所有结果进行流式处理。 connect.timeout 否 30s Duration 连接器在尝试连接到 MySQL 数据库服务器后超时前应等待的最长时间。 connect.max-retries 否 3 Integer 连接器应重试以建立 MySQL 数据库服务器连接的最大重试次数。 connection.pool.size 否 20 Integer 连接池大小。 jdbc.properties.* 否 无 String 传递自定义 JDBC URL 属性的选项。 用户可以传递自定义属性，如 'jdbc.properties.useSSL' = 'false'. heartbeat.interval 否 30s Duration 用于跟踪最新可用 binlog 偏移的发送心跳事件的间隔。 debezium.* 否 无 String 将 Debezium 的属性传递给 Debezium 嵌入式引擎，该引擎用于从 MySQL 服务器捕获数据更改。 例如: 'debezium.snapshot.mode' = 'never'. 查看更多关于 Debezium 的 MySQL 连接器属性 scan.incremental.close-idle-reader.enabled 否 false Boolean 是否在快照结束后关闭空闲的 Reader。 此特性需要'execution.checkpointing.checkpoints-after-tasks-finish.enabled' 需要设置为 true。

支持特性 增量快照读取 增量快照读取是一种读取表快照的新机制。与旧的快照机制相比，增量快照具有许多优点，包括： 在快照读取期间，Source 支持并发读取， 在快照读取期间，Source 支持进行 chunk 粒度的 checkpoint， 在快照读取之前，Source 不需要数据库锁权限。 如果希望 source 并行运行，则每个并行 reader 都应该具有唯一的 server id，因此server id的范围必须类似于 5400-6400， 且范围必须大于并行度。在增量快照读取过程中，MySQL CDC Source 首先通过表的主键将表划分成多个块（chunk）， 然后 MySQL CDC Source 将多个块分配给多个 reader 以并行读取表的数据。 无锁算法 MySQL CDC source 使用 增量快照算法, 避免了数据库锁的使用，因此不需要 “RELOAD” 权限。 并发读取 增量快照读取提供了并行读取快照数据的能力。 全量阶段支持 checkpoint 增量快照读取提供了在区块级别执行检查点的能力。它使用新的快照读取机制解决了以前版本中的检查点超时问题。

前提条件 MySQL CDC要求MySQL版本为5.6，5.7或8.0.x。 with参数中字段只能使用单引号，不能使用双引号。 该场景作业需要DLI与MySQL建立增强型跨源连接，且用户可以根据实际所需设置相应安全组规则。 如何建立增强型跨源连接，请参考《数据湖探索用户指南》中增强型跨源连接章节。 如何设置安全组规则，请参见《虚拟私有云用户指南》中“安全组”章节。 MySQL已开启了Binlog，并且binlog_row_image设置为FULL。 已创建MySQL用户，并授予了SELECT、 SHOW DATABASES 、REPLICATION SLAVE和REPLICATION CLIENT权限。注意: 在 scan.incremental.snapshot.enabled 参数已启用时（默认情况下已启用）时，不再需要授予 reload 权限。 GRANT SELECT, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'user' IDENTIFIED BY 'password';

GROUPING SETS Grouping Sets 可以通过一个标准的 GROUP BY 语句来描述更复杂的分组操作。数据按每个指定的 Grouping Sets 分别分组，并像简单的 group by 子句一样为每个组进行聚合。 GROUPING SETS 的每个子列表可以是：空的，多列或表达式，它们的解释方式和直接使用 GROUP BY 子句是一样的。一个空的 Grouping Sets 表示所有行都聚合在一个分组下，即使没有数据，也会输出结果。 对于 Grouping Sets 中的空子列表，结果数据中的分组或表达式列会用NULL代替。 SELECT supplier_id, rating, COUNT(*) AS total FROM (VALUES ('supplier1', 'product1', 4), ('supplier1', 'product2', 3), ('supplier2', 'product3', 3), ('supplier2', 'product4', 4)) AS Products(supplier_id, product_id, rating) GROUP BY GROUPING SETS ((supplier_id, rating), (supplier_id), ())

ROLLUP ROLLUP 是一种特定通用类型 Grouping Sets 的简写。代表着指定表达式和所有前缀的列表，包括空列表。 SELECT supplier_id, rating, COUNT(*) FROM (VALUES ('supplier1', 'product1', 4), ('supplier1', 'product2', 3), ('supplier2', 'product3', 3), ('supplier2', 'product4', 4)) AS Products(supplier_id, product_id, rating) GROUP BY ROLLUP (supplier_id, rating)

CUBE CUBE 是一种特定通用类型 Grouping Sets 的简写。代表着指定列表以及所有可能的子集和幂集。 例如：下面两个查询是等效的。 SELECT supplier_id, rating, product_id, COUNT(*) FROM (VALUES ('supplier1', 'product1', 4), ('supplier1', 'product2', 3), ('supplier2', 'product3', 3), ('supplier2', 'product4', 4)) AS Products(supplier_id, product_id, rating) GROUP BY CUBE (supplier_id, rating, product_id) SELECT supplier_id, rating, product_id, COUNT(*) FROM (VALUES ('supplier1', 'product1', 4), ('supplier1', 'product2', 3), ('supplier2', 'product3', 3), ('supplier2', 'product4', 4)) AS Products(supplier_id, product_id, rating) GROUP BY GROUPING SET ( ( supplier_id, product_id, rating ), ( supplier_id, product_id ), ( supplier_id, rating ), ( supplier_id ), ( product_id, rating ), ( product_id ), ( rating ), ( ) )

语法支持类型 CHAR, VARCHAR, STRING, BOOLEAN, BINARY, VARBINARY, BYTES, DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE, DATE, TIME, TIMESTAMP, TIMESTAMP_LTZ, INTERVAL, ARRAY, MULTISET, MAP, ROW, RAW 父主题： SQL语法约束与定义