5.8. Kafka Connector

Overview

This connector allows the use of Apache Kafka topics as tables in Presto.Each message is presented as a row in Presto.

Topics can be live: rows will appear as data arrives and disappear assegments get dropped. This can result in strange behavior if accessing thesame table multiple times in a single query (e.g., performing a self join).

Note

Apache Kafka 0.8+ is supported although it is highly recommend to use 0.8.1 or later.

Configuration

To configure the Kafka connector, create a catalog properties fileetc/catalog/kafka.properties with the following contents,replacing the properties as appropriate:

connector.name=kafka
kafka.table-names=table1,table2
kafka.nodes=host1:port,host2:port

Multiple Kafka Clusters

You can have as many catalogs as you need, so if you have additionalKafka clusters, simply add another properties file to etc/catalogwith a different name (making sure it ends in .properties). Forexample, if you name the property file sales.properties, Prestowill create a catalog named sales using the configured connector.

Configuration Properties

The following configuration properties are available:

kafka.table-names

Comma-separated list of all tables provided by this catalog. A table namecan be unqualified (simple name) and will be put into the default schema(see below) or qualified with a schema name (<schema-name>.<table-name>).

For each table defined here, a table description file (see below) mayexist. If no table description file exists, the table name is used as thetopic name on Kafka and no data columns are mapped into the table. Thetable will still contain all internal columns (see below).

This property is required; there is no default and at least one table must be defined.

kafka.default-schema

Defines the schema which will contain all tables that were defined withouta qualifying schema name.

This property is optional; the default is default.

kafka.nodes

A comma separated list of hostname:port pairs for the Kafka data nodes.

This property is required; there is no default and at least one node must be defined.

Note

Presto must still be able to connect to all nodes of the clustereven if only a subset is specified here as segment files may belocated only on a specific node.

kafka.connect-timeout

Timeout for connecting to a data node. A busy Kafka cluster may take quitesome time before accepting a connection; when seeing failed queries due totimeouts, increasing this value is a good strategy.

This property is optional; the default is 10 seconds (10s).

kafka.buffer-size

Size of the internal data buffer for reading data from Kafka. The databuffer must be able to hold at least one message and ideally can hold manymessages. There is one data buffer allocated per worker and data node.

This property is optional; the default is 64kb.

kafka.table-description-dir

References a folder within Presto deployment that holds one or more JSONfiles (must end with .json) which contain table description files.

This property is optional; the default is etc/kafka.

kafka.hide-internal-columns

In addition to the data columns defined in a table description file, theconnector maintains a number of additional columns for each table. Ifthese columns are hidden, they can still be used in queries but do notshow up in DESCRIBE <table-name> or SELECT *.

This property is optional; the default is true.

Internal Columns

For each defined table, the connector maintains the following columns:

For tables without a table definition file, the _key_corrupt and_message_corrupt columns will always be false.

Table Definition Files

Kafka maintains topics only as byte messages and leaves it to producersand consumers to define how a message should be interpreted. For Presto,this data must be mapped into columns to allow queries against the data.

Note

For textual topics that contain JSON data, it is entirely possible to notuse any table definition files, but instead use the PrestoJSON Functions and Operators to parse the _message column which containsthe bytes mapped into an UTF-8 string. This is, however, prettycumbersome and makes it difficult to write SQL queries.

A table definition file consists of a JSON definition for a table. Thename of the file can be arbitrary but must end in .json.

{
    "tableName": ...,
    "schemaName": ...,
    "topicName": ...,
    "key": {
        "dataFormat": ...,
        "fields": [
            ...
        ]
    },
    "message": {
        "dataFormat": ...,
        "fields": [
            ...
       ]
    }
}

Key and Message in Kafka

Starting with Kafka 0.8, each message in a topic can have an optional key.A table definition file contains sections for both key and message to mapthe data onto table columns.

Each of the key and message fields in the table definition is aJSON object that must contain two fields:

Each field definition is a JSON object:

{
    "name": ...,
    "type": ...,
    "dataFormat": ...,
    "mapping": ...,
    "formatHint": ...,
    "hidden": ...,
    "comment": ...
}

There is no limit on field descriptions for either key or message.

Row Decoding

For key and message, a decoder is used to map message and key data onto table columns.

The Kafka connector contains the following decoders:

• raw - Kafka message is not interpreted, ranges of raw message bytes are mapped to table columns
• csv - Kafka message is interpreted as comma separated message, and fields are mapped to table columns
• json - Kafka message is parsed as JSON and JSON fields are mapped to table columns
• avro - Kafka message is parsed based on an Avro schema and Avro fields are mapped to table columns

Note

If no table definition file exists for a table, the dummy decoder is used,which does not expose any columns.

raw Decoder

The raw decoder supports reading of raw (byte based) values from Kafka messageor key and converting it into Presto columns.

For fields, the following attributes are supported:

• dataFormat - selects the width of the data type converted
• type - Presto data type (see table below for list of supported data types)
• mapping - <start>[:<end>]; start and end position of bytes to convert (optional)The dataFormat attribute selects the number of bytes converted.If absent, BYTE is assumed. All values are signed.

Supported values are:

• BYTE - one byte
• SHORT - two bytes (big-endian)
• INT - four bytes (big-endian)
• LONG - eight bytes (big-endian)
• FLOAT - four bytes (IEEE 754 format)
• DOUBLE - eight bytes (IEEE 754 format)The type attribute defines the Presto data type on which the value is mapped.

Depending on Presto type assigned to column different values of dataFormat can be used:

The mapping attribute specifies the range of the bytes in a key ormessage used for decoding. It can be one or two numbers separated by a colon (<start>[:<end>]).

If only a start position is given:

• For fixed width types the column will use the appropriate number of bytes for the specified dateFormat (see above).
• When VARCHAR value is decoded all bytes from start position till the end of the message will be used.

If start and end position are given, then:

• For fixed width types the size must be equal to number of bytes used by specified dataFormat.
• For VARCHAR all bytes between start (inclusive) and end (exclusive) are used.

If no mapping attribute is specified it is equivalent to setting start position to 0 and leaving end position undefined.

Decoding scheme of numeric data types (BIGINT, INTEGER, SMALLINT, TINYINT, DOUBLE) is straightforward.A sequence of bytes is read from input message and decoded according to either:

• big-endian encoding (for integer types)
• IEEE 754 format for (for DOUBLE).

Length of decoded byte sequence is implied by the dataFormat.

For VARCHAR data type a sequence of bytes is interpreted according to UTF-8 encoding.

csv Decoder

The CSV decoder converts the bytes representing a message or key into astring using UTF-8 encoding and then interprets the result as a CSV(comma-separated value) line.

For fields, the type and mapping attributes must be defined:

• type - Presto data type (see table below for list of supported data types)
• mapping - the index of the field in the CSV recorddataFormat and formatHint are not supported and must be omitted.

Table below lists supported Presto types which can be used in type and decoding scheme:

json Decoder

The JSON decoder converts the bytes representing a message or key into aJSON according to RFC 4627. Note that the message or key MUST convertinto a JSON object, not an array or simple type.

For fields, the following attributes are supported:

• type - Presto type of column.
• dataFormat - Field decoder to be used for column.
• mapping - slash-separated list of field names to select a field from the JSON object
• formatHint - only for custom-date-time, see belowThe JSON decoder supports multiple field decoders, with _default beingused for standard table columns and a number of decoders for date and timebased types.

Table below lists Presto data types which can be used as in type and matching field decoderswhich can be specified via dataFormat attribute

Default Field decoder

This is the standard field decoder supporting all the Presto physical datatypes. A field value will be coerced by JSON conversion rules intoboolean, long, double or string values. For non-date/time based columns,this decoder should be used.

Date and Time Decoders

To convert values from JSON objects into Presto DATE, TIME, TIME WITH TIME ZONE`,``TIMESTAMP or TIMESTAMP WITH TIME ZONE columns, special decoders must be selected using thedataFormat attribute of a field definition.

• iso8601 - text based, parses a text field as an ISO 8601 timestamp.
• rfc2822 - text based, parses a text field as an RFC 2822 timestamp.
• • custom-date-time - text based, parses a text field according to Joda format pattern
  • specified via formatHint attribute. Format pattern should conformto https://www.joda.org/joda-time/apidocs/org/joda/time/format/DateTimeFormat.html.
• milliseconds-since-epoch - number based, interprets a text or number as number of milliseconds since the epoch.
• seconds-since-epoch - number based, interprets a text or number as number of milliseconds since the epoch.For TIMESTAMP WITH TIME ZONE and TIME WITH TIME ZONE data types, if timezone information is present in decoded value, it willbe used in Presto value. Otherwise result time zone will be set to UTC.

avro Decoder

The Avro decoder converts the bytes representing a message or key inAvro format based on a schema. The message must have the Avro schema embedded.Presto does not support schema-less Avro decoding.

For key/message, using avro decoder, the dataSchema must be defined.This should point to the location of a valid Avro schema file of the message which needs to be decoded. This location can be a remote web server(e.g.: dataSchema: 'http://example.org/schema/avro_data.avsc') or local file system(e.g.: dataSchema: '/usr/local/schema/avro_data.avsc').The decoder will fail if this location is not accessible from the Presto coordinator node.

For fields, the following attributes are supported:

• name - Name of the column in the Presto table.
• type - Presto type of column.
• mapping - slash-separated list of field names to select a field from the Avro schema. If field specified in mapping does not exist in the original Avro schema then a read operation will return NULL.Table below lists supported Presto types which can be used in type for the equivalent Avro field type/s.

Avro schema evolution

The Avro decoder supports schema evolution feature with backward compatibility. With backward compatibility,a newer schema can be used to read Avro data created with an older schema. Any change in the Avro schema must also bereflected in Presto’s topic definition file. Newly added/renamed fields must have a default value in the Avro schema file.

The schema evolution behavior is as follows:

• Column added in new schema:Data created with an older schema will produce a default value when table is using the new schema.
• Column removed in new schema:Data created with an older schema will no longer output the data from the column that was removed.
• Column is renamed in the new schema:This is equivalent to removing the column and adding a new one, and data created with an older schemawill produce a default value when table is using the new schema.
• Changing type of column in the new schema:If the type coercion is supported by Avro, then the conversion happens. An error is thrown for incompatible types.