Avro

Specify the avro format when your Kafka message data is a single-object encoded Avro file or you are using the Confluent Schema Registry to load Avro message key and/or value data. gpkafka reads Avro data from Kafka and loads it into a single JSON-type column. You must define a mapping if you want gpkafka to write the data into specific columns in the target Greenplum Database table.

Binary

Use the binary format when your Kafka message data is a stream of bytes. gpkafka reads binary data from Kafka and loads it into a single bytea-type column.

CSV

Use the csv format when your Kafka message data is comma-delimited text and conforms to RFC 4180. The message content may not contain line ending characters (CR and LF).

Data in csv format may appear in Kafka messages as follows:

"1313131","12","backorder","1313.13"
"3535353","11","shipped","761.35"
"7979797","11","partial","18.72"

Custom

The Greenplum Streaming Server provides a custom data formatter plug-in framework for Kafka messages using user-defined functions. The type of Kafka message data supported by a custom formatter is formatter-specific. For example, a custom formatter may support compressed or complex data.

Delimited Text

The Greenplum Streaming Server supports loading Kafka data delimited by one or more characters that you specify. Use the delimited format for such data. The delimiter may be a multi-byte value and up to 32 bytes in length. You cannot specify a quote or an escape character in the delimiter.

Sample data using a pipe ('|') delimiter character follows:

1313131|12|backorder|1313.13
3535353|11|shipped|761.35
7979797|11|partial|18.72

JSON

Specify the json format when your Kafka message data is in JSON or JSONB format. gpkafka reads JSON data from Kafka only as a single column. You must define a mapping if you want gpkafka to write the data into specific columns in the target Greenplum Database table.

Sample JSON message data:

{ "cust_id": 1313131, "month": 12, "amount_paid":1313.13 }
{ "cust_id": 3535353, "month": 11, "amount_paid":761.35 }
{ "cust_id": 7979797, "month": 11, "amount_paid":18.82 }

Greenplum Database Options (Version 2-Focused)

You identify the Greenplum Database connection options via the DATABASE, USER, PASSWORD, HOST, and PORT parameters.

The VERSION parameter identifies the version of the gpkafka YAML configuration file. The default version is Version 1.

KAFKA:INPUT Options

Specify the Kafka brokers and topic of interest using the SOURCE block. You must create the Kafka topic prior to loading data. By default, GPSS reads Kafka messages from all partitions. You may specify a single, a comma-separated list, and/or a range of partition numbers to restrict the partitions from which GPSS reads messages. The PARTITIONS property is supported only for version 2 and 3 load configuration file formats.

When you provide a VALUE block, you must specify the COLUMNS and FORMAT parameters. The VALUE:COLUMNS block includes the name and type of each data element in the Kafka message. The default source-to-target data mapping behaviour of gpkafka is to match a column name as defined in COLUMNS:NAME with a column name in the target Greenplum Database OUTPUT:TABLE:

• You must identify the Kafka data elements in the order in which they appear in the Kafka message.
• You may specify NAME: __IGNORED__ to omit a Kafka message value data element from the load operation.
• You must provide the same name for each non-ignored Kafka data element and its associated Greenplum Database table column.
• You must specify an equivalent data type for each non-ignored Kafka data element and its associated Greenplum Database table column.

The VALUE:FORMAT keyword identifies the format of the Kafka message value. gpkafka supports comma-delimited text format (csv ) and data that is separated by a configurable delimiter (delimited). gpkafka also supports binary (binary), JSON/JSONB (json), custom (custom), and Avro (avro) format value data.

When you provide a KEY block, you must specify the COLUMNS and FORMAT parameters. The KEY:COLUMNS block includes the name and type of each element of the Kafka message key, and is subject to the same restrictions as identified for VALUE:COLUMNS above. The KEY:FORMAT keyword identifies the format of the Kafka message key. gpkafka supports avro, binary, csv, custom, delimited, and json format key data.

The FILTER parameter identifies a filter to apply to the Kafka input messages before the data is loaded into Greenplum Database. If the filter evaluates to true, gpkafka loads the message. The message is dropped if the filter evaluates to false. The filter string must be a valid SQL conditional expression and may reference one or more KEY or VALUE column names.

The ERROR_LIMIT parameter identifies the number of errors or the error percentage threshold after which gpkafka should exit the load operation. The default ERROR_LIMIT is zero; the load operation is stopped when the first error is encountered.

KAFKA:OUTPUT Options

You identify the target Greenplum Database schema name and table name via the KAFKA:OUTPUT: SCHEMA and TABLE parameters. You must pre-create the Greenplum Database table before you attempt to load Kafka data.

The default load mode is to insert Kafka data into the Greenplum Database table. gpkafka also supports updating and merging Kafka message data into a Greenplum table. You specify the load MODE, the MATCH_COLUMNS and UPDATE_COLUMNS, and any UPDATE_CONDITIONs that must be met to merge or update the data. In MERGE MODE, you can also specify ORDER_COLUMNS to filter out duplicates and a DELETE_CONDITION.

About the Merge Load Mode

MERGE mode is similar to an UPSERT operation; gpkafka may insert new rows in the database, or may update an existing database row that satisfies match and update conditions. gpkafka deletes rows in MERGE mode when the data satisfies an optional DELETE_CONDITION that you specify.

gpkafka stages a merge operation in a temporary table, generating the SQL to populate the temp table from the set of OUTPUT configuration properties that you provide.

gpkafka uses the following algorithm for MERGE mode processing:

1. Create a temporary table like the target table.
2. Generate the SQL to insert the source data into the temporary table.
   1. Add the MAPPINGS.
   2. Add the FILTER.
   3. Use MATCH_COLUMNS and ORDER_COLUMNS to filter out duplicates.
3. Update the target table from rows in the temporary table that satisfy MATCH_COLUMNS, UPDATE_COLUMNS, and UPDATE_CONDITION.
4. Insert non-matching rows into the target table.
5. Delete rows in the target table that satisfy MATCH_COLUMNS and the DELETE_CONDITION.
6. Truncate the temporary table.

Other Options

The KAFKA:METADATA:SCHEMA parameter specifies the name of the Greenplum Database schema in which gpkafka creates external and history tables.

gpkafka commits Kafka data to the Greenplum Database table at the row and/or time intervals that you specify in the KAFKA:COMMIT: MAX_ROW and/or MINIMAL_INTERVAL parameters. You must specify at least one of these parameters.

You can configure gpkafka to execute a task (user-defined function or SQL commands) after GPSS reads a configurable number of batches from Kafka. Use the KAFKA:TASK: POST_BATCH_SQL and BATCH_INTERVAL configuration parameters to specify the task and the batch interval.

Specify a KAFKA:PROPERTIES block to set Kafka consumer configuration properties. gpkafka sends the property names and values to Kafka when it instantiates a consumer for the load operation.

About KEYs, VALUEs, and FORMATs

You can specify any data format in the Version 2 configuration file KEY:FORMAT and VALUE:FORMAT parameters, with some restrictions. The Greenplum Streaming Server supports the following KEY:FORMAT and VALUE:FORMAT combinations:

About Transforming and Mapping Kafka Input Data

You can define a MAPPING between the Kafka input data (VALUE:COLUMNS, KEY:COLUMNS, and META:COLUMNS) and the columns in the target Greenplum Database table. Defining a mapping may be useful when you have a multi-field input column (such as a JSON-type column), and you want to assign individual components of the input field to specific columns in the target table.

You might also use a MAPPING to assign a value expression to a target table column. The expression must be one that you could specify in the SELECT list of a query, and can include a constant value, a column reference, an operator invocation, a built-in or user-defined function call, and so forth.

If you choose to map more than one input column in an expression, you can can create a user-defined function to parse and transform the input column and return the columns of interest.

For example, suppose a Kafka producer emits the following JSON messages to a topic:

{ "customer_id": 1313131, "some_intfield": 12 }
{ "customer_id": 77, "some_intfield": 7 }
{ "customer_id": 1234, "some_intfield": 56 }

You could define a user-defined function, udf_parse_json(), to parse the data as follows:

=> CREATE OR REPLACE FUNCTION udf_parse_json(value json)
     RETURNS TABLE (x int, y text)
   LANGUAGE plpgsql AS $$
     BEGIN
        RETURN query
        SELECT ((value->>'customer_id')::int), ((value->>'some_intfield')::text);
     END $$;

This function returns the two fields in each JSON record, casting the fields to integer and text, respectively.

An example MAPPING for the topic data in a JSON-type KAFKA:INPUT:COLUMNS named jdata follows:

MAPPING: 
  cust_id: (jdata->>'customer_id')
  field2: ((jdata->>'some_intfield') * .075)::decimal
  j1, j2: (udf_parse_json(jdata)).*

The Greenplum Database table definition for this example scenario is:

=> CREATE TABLE t1map( cust_id int, field2 decimal(7,2), j1 int, j2 text );

About Mapping Avro Bytes Fields to Base64-Encoded Strings

When you specify AVRO_OPTION:BYTES_TO_BASE64, GPSS maps Avro bytes fields to base64-encoded strings. You can provide a MAPPING to decode these strings and write the data to a Greenplum bytea column.

For example, if the Avro schema is:

{
    "type": "record",
    "name": "bytes_test",
    "fields": [
                {"name": "id", "type": "long"},
                {"name": "string", "type": "string"},
                {"name": "bytes",  "type": "bytes"},
        {
            "name": "inner_record",
            "type": {
                "type": "map",
                "values": {
                                                "type": "bytes",
                                                "name": "nested_bytes"
                }
            }
        }
    ]
}

And if your load configuration file includes these input property settings:

VALUE:
    COLUMNS:
      - NAME: c1
        TYPE: json
    FORMAT: avro
    AVRO_OPTION:
      SCHEMA_REGISTRY_ADDR: http://localhost:8081
      BYTES_TO_BASE64: true

You can define a MAPPING to decode the encoded strings as follows:

MAPPING:
  - NAME: id
    EXPRESSION: (c1->>'id')::int
  - NAME: bytes1
    EXPRESSION: (decode(c1->>'bytes', 'base64'))
  - NAME: bytes2
    EXPRESSION: (decode((c1->>'inner_record')::json->>'nested_bytes', 'base64'))

This mapping decodes the bytes1 and bytes2 fields to the Greenplum bytea data type. GPSS would expect to load these mapped fields to a Greenplum table with the following definition:

CREATE TABLE avbyte( id int, bytes1 bytea, bytes2 bytea);