Vertica Sink Connector 使用必看手册

@

点击蓝字

关注我们

Spark
Flink
SeaTunnel Zeta

使用 Xa transactions
 保证 精确一次性处理
. 所以 精确一次性处理
 仅可用于支持Xa transactions
的数据库. 可以设置 is_exactly_once=true
来使用.

下载与'Maven'对应的支持列表，并将其复制到'$SEATNUNNEL_HOME/plugins/jdbc/lib/'工作目录。

例如 Vertica datasource: cp vertica-jdbc-xxx.jar $SEATNUNNEL_HOME/plugins/jdbc/lib/

Tips

如果未设置 partition_column
，将以单并发方式运行；如果设置了 partition_column
，将根据任务的并发度并行执行。

该示例定义了一个 SeaTunnel 同步任务，它通过 FakeSource 自动生成数据并将其发送到 JDBC Sink。FakeSource 生成了总共 16 行数据（row.num=16），每行有两个字段，name（字符串类型）和 age（整数类型）。最终目标表是 test_table，表中也会有 16 行数据。在运行此作业之前，您需要在您的 Vertica 数据库中创建数据库 test 和表 test_table。如果您尚未安装和部署 SeaTunnel，请按照安装 SeaTunnel中的说明进行安装和部署。然后按照SeaTunnel Engine 快速入门(https://chat.openai.com/start-v2/locally/quick-start-seatunnel-engine.md)中的说明运行此作业。

# Defining the runtime environment
env {
  # You can set flink configuration here
  execution.parallelism = 1
  job.mode = "BATCH"
}

source {
  # This is a example source plugin **only for test and demonstrate the feature source plugin**
  FakeSource {
    parallelism = 1
    result_table_name = "fake"
    row.num = 16
    schema = {
      fields {
        name = "string"
        age = "int"
      }
    }
  }
  # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
  # please go to https://seatunnel.apache.org/docs/category/source-v2
}

transform {
  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
    # please go to https://seatunnel.apache.org/docs/category/transform-v2
}

sink {
    jdbc {
        url = "jdbc:vertica://localhost:5433/vertica"
        driver = "com.vertica.jdbc.Driver"
        user = "root"
        password = "123456"
        query = "insert into test_table(name,age) values(?,?)"
        }
  # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
  # please go to https://seatunnel.apache.org/docs/category/sink-v2
}

这个示例不需要编写复杂的SQL语句，您可以配置数据库名称和表名称，以自动为您生成添加语句。

sink {
    jdbc {
        url = "jdbc:vertica://localhost:5433/vertica"
        driver = "com.vertica.jdbc.Driver"
        user = "root"
        password = "123456"
        # Automatically generate sql statements based on database table names
        generate_sink_sql = true
        database = test
        table = test_table
    }
}

对于精确写入场景，我们保证精确一次性。

sink {
    jdbc {
        url = "jdbc:vertica://localhost:5433/vertica"
        driver = "com.vertica.jdbc.Driver"
    
        max_retries = 0
        user = "root"
        password = "123456"
        query = "insert into test_table(name,age) values(?,?)"
    
        is_exactly_once = "true"
    
        xa_data_source_class_name = "com.vertical.cj.jdbc.VerticalXADataSource"
    }
}

一键三连-点赞在看转发⭐️！