Synopsis

DATABASE: db_name
USER: user_name
PASSWORD: password
HOST: host
PORT: greenplum_port
[VERSION: 1]
KAFKA:
   INPUT:
      SOURCE:
        BROKERS: kafka_broker_host:broker_port [, ... ]
        TOPIC: kafka_topic
      [COLUMNS:
         - NAME: { column_name | __IGNORED__ }
           TYPE: column_data_type
         [ ... ]]
      FORMAT: data_format
      [[DELIMITED_OPTION:
         DELIMITER: delimiter_string] |
      [AVRO_OPTION:
         SCHEMA_REGISTRY_ADDR: http://schemareg_host:schemareg_port [, ... ]] |
      [CUSTOM_OPTION:
         NAME: udf_name
         PARAMSTR: udf_parameter_string]]
      [FILTER: filter_string]
      ERROR_LIMIT: { num_errors | percentage_errors }
   OUTPUT:
      [SCHEMA: output_schema_name]
      TABLE: table_name
      [MODE: mode]
      [MATCH_COLUMNS:
         - match_column_name
         [ ... ]]
      [UPDATE_COLUMNS:
         - update_column_name
         [ ... ]]
      [UPDATE_CONDITION: update_condition]
      [MAPPING:
         - NAME: target_column_name
           EXPRESSION: { source_column_name | expression }
         [ ... ]
           |
         target_column_name : { source_column_name | expression }
         [ ... ] ]
   [METADATA:
      [SCHEMA: metadata_schema_name]]
   COMMIT:
      MAX_ROW: num_rows
      MINIMAL_INTERVAL: wait_time
   [POLL:
      BATCHSIZE: num_records
      TIMEOUT: poll_time]
   [TASK:
      POST_BATCH_SQL: udf_or_sql_to_run
      BATCH_INTERVAL: num_batches]
    [PROPERTIES:
      kafka_property_name: kafka_property_value
      [ ... ]]

Description

You specify load configuration parameters for the gpkafka utilities in a YAML-formatted configuration file. (This reference page uses the name gpkafka.yaml when referring to this file; you may choose your own name for the file.) Load parameters include Greenplum Database connection and target table information, Kafka broker and topic information, and error and commit thresholds.

The gpkafka 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, and keywords are 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 gpkafka.yaml configuration file. For example, if you create a table as follows:

CREATE TABLE "MyTable" ("MyColumn" text);

Your gpkafka.yaml YAML configuration file would refer to the above table and column names as:

  COLUMNS:
     - name: '"MyColumn"'
       type: text
OUTPUT:
   TABLE: '"MyTable"'

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

Examples

Load data from Kafka as defined in the Version 1 configuration file named kafka2greenplum.yaml:

gpkafka load kafka2greenplum.yaml

Example kafka2greenplum.yaml configuration file:

DATABASE: ops
USER: gpadmin
HOST: mdw-1
PORT: 5432
KAFKA:
   INPUT:
      SOURCE:
         BROKERS: kbrokerhost1:9092
         TOPIC: customer_expenses
      COLUMNS:
         - NAME: cust_id
           TYPE: int
         - NAME: month
           TYPE: int
         - NAME: expenses
           TYPE: decimal(9,2)
      FORMAT: delimited
      DELIMITED_OPTION:
         DELIMITER: '|'
      ERROR_LIMIT: 25
   OUTPUT:
      SCHEMA: payables
      TABLE: expenses
      MAPPING:
        - NAME: customer_id 
          EXPRESSION: cust_id
        - NAME: newcust
          EXPRESSION: cust_id > 5000000
        - NAME: expenses 
          EXPRESSION: expenses
        - NAME: tax_due 
          EXPRESSION: expenses * .0725
   METADATA:
      SCHEMA: gpkafka_internal
   COMMIT:
      MINIMAL_INTERVAL: 2000

See Also

gpkafka-v2.yaml, gpkafka history, gpkafka load, gpss, gpss.json