Clickzetta Spark Flink Connector

ClickZetta Spark & Flink Connector

Read references/spark.md for the Spark Connector and references/flink.md for the Flink Write Connector.

Key Constraints (Required Reading)

Spark Connector Quick Example

// Write
df.write.format("clickzetta")
  .option("endpoint", "your_instance.cn-shanghai-alicloud.api.clickzetta.com")
  .option("username", sys.env("CZ_USERNAME"))
  .option("password", sys.env("CZ_PASSWORD"))
  .option("workspace", "your_workspace")
  .option("virtualCluster", "default")
  .option("schema", "public")
  .option("table", "orders")
  .mode("append")
  .save()

// Read
val df = spark.read.format("clickzetta")
  .option("endpoint", "your_instance.cn-shanghai-alicloud.api.clickzetta.com")
  .option("username", sys.env("CZ_USERNAME"))
  .option("password", sys.env("CZ_PASSWORD"))
  .option("workspace", "your_workspace")
  .option("virtualCluster", "default")
  .option("schema", "public")
  .option("table", "orders")
  .load()

Flink Connector Quick Example

-- CDC mode (supports INSERT/UPDATE/DELETE; the target table must have a primary key)
CREATE TABLE lakehouse_sink (
    order_id   INT,
    status     STRING,
    amount     DOUBLE,
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'connector'       = 'igs-dynamic-table',
    'curl'            = 'jdbc:clickzetta://your_instance.cn-shanghai-alicloud.api.clickzetta.com/default?username=user&password=***&schema=public',
    'schema-name'     = 'public',
    'table-name'      = 'orders',
    'sink.parallelism' = '1',
    'properties'      = 'authentication:true'
);

INSERT INTO lakehouse_sink SELECT order_id, status, amount FROM source_table;

Selection Guide