Synopsis

DATABASE: db_name
USER: user_name
PASSWORD: password
HOST: host
PORT: greenplum_port
VERSION: 2
KAFKA:
   INPUT:
      SOURCE:
        BROKERS: kafka_broker_host:broker_port [, ... ]
        TOPIC: kafka_topic
        ENCRYPTION: bool_value
      [VALUE:
        COLUMNS:
           - NAME: { column_name | __IGNORED__ }
             TYPE: column_data_type
           [ ... ]
         FORMAT: value_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]]
      [KEY:
        COLUMNS:
           - NAME: { column_name | __IGNORED__ }
             TYPE: column_data_type
           [ ... ]
         FORMAT: key_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 }
      [LOCAL_HOSTNAME: local_hostname]
      [LOCAL_PORT: local_port]
   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' }
         [ ... ]]
   [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 2 configuration file named kafka2greenplumv2.yaml:

gpkafka load kafka2greenplumv2.yaml

Example kafka2greenplumv2.yaml configuration file:

DATABASE: ops
USER: gpadmin
HOST: mdw-1
PORT: 5432
VERSION: 2
KAFKA:
   INPUT:
      SOURCE:
         BROKERS: kbrokerhost1:9092
         TOPIC: customer_expenses2
      VALUE:
         COLUMNS:
           - NAME: c1
             TYPE: json
         FORMAT: avro
         AVRO_OPTION:
           SCHEMA_REGISTRY_ADDR: http://localhost:8081
      KEY:
         COLUMNS:
           - NAME: key
             TYPE: json
         FORMAT: avro
         AVRO_OPTION:
           SCHEMA_REGISTRY_ADDR: http://localhost:8081
      ERROR_LIMIT: 25
   OUTPUT:
      SCHEMA: payables
      TABLE: expenses2
      MAPPING:
        - NAME: customer_id
          EXPRESSION: (c1->>'cust_id')::int
        - NAME: newcust
          EXPRESSION: ((c1->>'cust_id')::int > 5000000)::boolean
        - NAME: expenses
          EXPRESSION: (c1->>'expenses')::decimal
        - NAME: tax_due
          EXPRESSION: ((c1->>'expenses')::decimal * .075)::decimal
   METADATA:
      SCHEMA: gpkafka_internal
   COMMIT:
      MAX_ROW: 1000
      MINIMAL_INTERVAL: 30000

See Also

Example: Loading Avro Data from Kafka into Greenplum, gpkafka.yaml, gpkafka load, gpkafka check, gpss, gpss.json