Databricks数据源

使用说明

• 使用限制

  • 资源组：Databricks Reader 仅支持 Serverless资源组，并且需要为资源组绑定的VPC配置公网 NAT 网关和EIP。

  • 读取模式：数据同步仅支持 JDBC模式读取数据。

• 并发读取与数据一致性问题

  Databricks Reader 通过 splitPk 配置对数据进行切分，并启动多个并发任务以提高同步效率。但请注意以下关键点：

  • 多个并发任务不属于同一个数据库事务，且存在时间间隔。

  • 在数据源持续写入的情况下，并发读取可能导致获取的数据快照不完整或不一致。

  解决方案：由于技术限制，无法实现跨多线程的完美一致性快照。我们提供以下两种工程上的解决方案，请根据业务场景权衡选择：

  • 方案一：配置为单线程同步，不设置数据切片（splitPk）。可以保证数据严格一致，但同步速度较慢。

  • 方案二：在同步期间确保数据源为静态。例如，通过锁表、临时关闭应用写入或暂停备库同步等方式。同步速度较快，但可能对线上业务造成影响。

• 编码处理

  Databricks Reader 底层使用 JDBC 进行数据抽取，JDBC 会自动处理不同数据库的编码识别与转换。因此，您无需手动配置编码。

• 增量数据同步

  Databricks Reader 通过执行 SELECT ... WHERE ... 语句来抽取数据，实现增量同步的关键在于 WHERE 条件的构造。

  1. 推荐方式（基于时间戳字段）：

     1. 在您的源数据库表中，设计一个时间戳字段（例如 modify_time）。

     2. 当应用对数据进行新增或更新（包括逻辑删除）时，务必同步更新该时间戳字段。

     3. 在同步任务中，使用该时间戳字段作为 WHERE 条件，即可拉取上一同步点之后发生变化的数据。

  2. 不支持的情况：如果业务表中没有可用于区分新增或修改的字段（如时间戳或自增ID），Databricks Reader 无法进行增量同步，只能进行全量数据同步。

支持的字段类型

在进行离线读取时，Databricks Reader支持大部分Databricks类型，但也存在个别类型未被支持的情况，请注意检查您的数据类型。

Databricks Reader针对Databricks类型的转换列表，如下所示。

创建数据源

在进行数据同步任务开发时，您需要在DataWorks上创建一个对应的数据源，操作流程请参见数据源管理，详细的配置参数解释可在配置界面查看对应参数的文案提示。

数据同步任务开发

数据同步任务的配置入口和通用配置流程可参见下文的配置指导。

常见问题（FAQ）

• Q：读取报错：[Databricks][JDBCDriver](500313) Error getting the data value from result set: Column13:[Databricks][JDBCDriver](500312) Error in fetching data rows: Timestamp Conversion has failed.

  A：Databricks 的 TIMESTAMP 类型取值范围超过 JAVA Timestamp 取值范围，如果出现了超出范围的值，JDBC Driver 会报错。可以通过修改 column 配置解决，例如"column": ["CAST(col_timestamp AS STRING)"] 。

附录：脚本Demo与参数说明

{
  "type": "job",
  "version": "2.0",
  "steps": [
    {
      "stepType": "databricks",
      "parameter": {
        "datasource": "databricks",
        "schema": "schema1",
        "table": "table1",
        "readMode": "jdbc",
        "where": "id>1",
        "splitPk": "id",
        "column": [
          "c1",
          "c2"
        ]
      },
      "name": "Reader",
      "category": "reader"
    },
    {
      "stepType": "stream",
      "parameter": {},
      "name": "Writer",
      "category": "writer"
    }
  ],
  "setting": {
    "errorLimit": {
      "record": "0"
    },
    "speed": {
      "concurrent": 1
    }
  }
}