EINBLIQ.IO

Kafka

If your CDN logs already flow through Kafka, a Kafka Connect S3 sink writes them to a bucket we manage.

Last checked

Values you receive from us

Value Used for
<BUCKET_NAME> Amazon S3 bucket
<BUCKET_PREFIX> Prefix for your files inside the bucket
<AWS_REGION> Bucket region
<ACCESS_KEY_ID> Access key ID for the bucket
<SECRET_ACCESS_KEY> Secret access key

Before you start

  • A running Kafka Connect cluster with access to the topic that holds your CDN log lines.
  • Log lines as flat JSON records, one request per record, with the fields listed below.
  • The Amazon S3 Sink Connector by Confluent, version 12.1.15 at the time of writing, installed on your Connect workers:
Shell
confluent connect plugin install confluentinc/kafka-connect-s3:latest

The Confluent connector is distributed under the Confluent Community License. If you need an Apache-2.0 licensed connector or CSV output, the Aiven S3 sink connector supports JSON, JSON Lines, CSV and Parquet; we adapt the setup with you.

Connector configuration

Save the configuration as einbliq-s3-sink.json. It writes gzip-compressed, newline-delimited JSON, partitioned by hour.

JSON
{
  "name": "einbliq-s3-sink",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "tasks.max": "1",
    "topics": "<CDN_LOG_TOPIC>",
    "s3.bucket.name": "<BUCKET_NAME>",
    "s3.region": "<AWS_REGION>",
    "aws.access.key.id": "<ACCESS_KEY_ID>",
    "aws.secret.access.key": "<SECRET_ACCESS_KEY>",
    "topics.dir": "<BUCKET_PREFIX>",
    "storage.class": "io.confluent.connect.s3.storage.S3Storage",
    "format.class": "io.confluent.connect.s3.format.json.JsonFormat",
    "s3.compression.type": "gzip",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
    "partition.duration.ms": "3600000",
    "path.format": "'year'=YYYY/'month'=MM/'day'=dd/'hour'=HH",
    "locale": "en-US",
    "timezone": "UTC",
    "timestamp.extractor": "Record",
    "flush.size": "10000",
    "rotate.interval.ms": "300000"
  }
}

<CDN_LOG_TOPIC> is the name of your topic. flush.size commits a file after 10,000 records. With timestamp.extractor: Record, rotate.interval.ms advances with record timestamps; an idle topic can delay a commit until another record arrives. If you need rotation based on wall-clock time, agree a rotate.schedule.interval.ms configuration with us. See the Confluent rotation settings.

Deploy the connector

Create the connector through the Kafka Connect REST API:

Shell
curl -X POST "http://<CONNECT_HOST>:8083/connectors" \
  -H "Content-Type: application/json" \
  --data @einbliq-s3-sink.json

Check its state with GET /connectors/einbliq-s3-sink/status. Both the connector and its tasks should report RUNNING.

Required fields

Each record needs the values listed in Log fields we need. Field names can follow your CDN; we map them during onboarding.

JSON
{
  "hostname": "example.com",
  "clientIP": "192.0.2.10",
  "edgeIP": "203.0.113.1",
  "userAgent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
  "timestamp": 1704110400000,
  "statusCode": 200,
  "segmentURI": "/video/segment_001.m4s",
  "payloadSize": 524288,
  "duration": 150,
  "cacheStatus": "HIT",
  "queryString": "CMCD=bl%3D21300%2Cbr%3D3200%2Cd%3D4004%2Cot%3Dv%2Csid%3D%226e2fb550-c457-11e9-bb97-0800200c9a66%22",
  "referer": "https://example.com/watch",
  "cmcd": "bl=21300,br=3200,d=4004,ot=v,sid=\"6e2fb550-c457-11e9-bb97-0800200c9a66\""
}

If your pipeline stores CMCD only in the query string, the separate cmcd field can be omitted.

Check the delivery

Files appear in the bucket after the first rotation interval. Tell us when the connector runs; we confirm that the records contain all fields and CMCD.