Synopsis

target:
  host: host
  port: greenplum_port
  user: user_name
  password: password
  database: db_name
  schema: schema_name
  table: table_name

source:
  kafka:
    topic: kafka_topic
    brokers: kafka_broker_host:broker_port %, ...%
    partitions: (partition_numbers)
    fallback_offset: earliest | latest
    key_content:
      data_format:
        column_spec
        other_props
    value_content:
      data_format:
        column_spec
        other_props
    meta:
      json:
        column:
          name: meta
          type: json
    rdkafka_prop:
      kafka_property_name: kafka_property_value
      ... 

channel:
  gpdb_channel:
    mode:
      # specify a single mode property block (described below)
      insert:
        mode_specific_property: value
        ...
      update:
        mode_specific_property: value
        ...
      merge:
        mode_specific_property: value
        ...
    work_schema: work_schema_name
    error_limit: num_errors | percentage_errors
    window:
      batch:
        max_count: number_of_rows
        interval_ms: wait_time
        idle_duration_ms: idle_time
      window_size: num_batches
      window_statement: udf_or_sql_to_run
    mapping:
      target_column_name : source_column_name | expression
      ...

avro:
  source_column_name: column_name
  schema_url: http://schemareg_host:schemareg_port %, ...%
  bytes_to_base64: boolean

binary:
  source_column_name: column_name

csv:
  columns:
    - name: column_name
      type: column_data_type
    ...

custom:
  columns:
    - name: column_name
      type: column_data_type
    ...
  name: formatter_name
  options:
    - optname=optvalue
    ...

delimited:
  columns:
    - name: column_name
      type: column_data_type
    ...
  delimiter: delimiter_string

json:
  column:
    name: column_name
    type: json | jsonb

Where the mode_specific_propertys that you can specify for each mode follow:

insert:
  filter_expression: filter_string

update:
  filter_expression: filter_string
  match_columns: [match_column_names]
  order_columns: [order_column_names]
  update_columns: [update_column_names]
  update_condition: update_condition

merge:
  filter_expression: filter_string
  match_columns: [match_column_names]
  update_columns: [update_column_names]
  order_columns: [order_column_names]
  update_condition: update_condition
  delete_condition: delete_condition

Description

You specify load configuration properties for a Greenplum Streaming Server (GPSS) Kafka load job in a YAML-formatted configuration file. (This reference page uses the name gpkafka-v3.yaml when referring to this file; you may choose your own name for the file.) Load properties include Greenplum Database connection and data import properties, Kafka broker, topic, and message format information, and properies specific to the GPSS job.

The gpkafka load utility processes the YAML configuration file in order, using indentation (spaces) to determine the document hierarchy and the relationships between the sections. The use of white space in the file is significant. Keywords are not case-sensitive.

Keywords and Values

Notes

If you created a database object name using a double-quoted identifier (delimited identifier), you must specify the delimited name within single quotes in the load configuration file. For example, if you create a table as follows:

CREATE TABLE "MyTable" (c1 text);

Your YAML configuration file would refer to the table name as:

target:
   table: '"MyTable"'

Kafka Properties

PROPERTIES:
      api.version.request: false
      broker.version.fallback: 0.8.2.1

Examples

Load data from Kafka as defined in the Version 3 configuration file named kafka2greenplumv3.yaml:

gpkafka load kafka2greenplumv3.yaml

Example kafka2greenplumv3.yaml configuration file:

target:
  host: mdw-1
  port: 15432
  user: gpadmin
  password: changeme
  database: testdb
  schema: public
  table: tbl_order_merge
source:
  kafka:
    topic: daily_orders
    brokers: localhost:9092
    key_content:
      binary:
        source_column_name: key
    value_content:
      json:
        column:
          name: value
          type: JSON
    meta:
      json:
        column:
          name: meta
          type: JSON
channel:
  gpdb_channel:
    mode:
      merge:
        match_columns: [pk]
        update_columns: [data, key]
        order_columns: [p, o]
    work_schema: public
    error_limit: "25"
    batch_window:
      batch:
        interval_ms: 5000
    mapping:
      data: (value->>'data')::text
      o: (meta->>'offset')::bigint
      p: (meta->>'partition')::int
      pk: (value->>'pk')::int

See Also

gpkafka.yaml, gpkafka load