This application can run in either service
or worker
modes. The service
consumes JSON bodies from HTTP, converts them into Firehose format and writes them to a Kafka topic. The worker
consumes the Kafka topic, aggregates events and writes groups of them to an ORC file in S3.
The application produces a docker image which determines it's mode based on the first argument.
docker build -t shokuyoku:test .
docker run shokuyoku:test service
You should add environment variables for proper execution.
KAFKA_TOPIC environment variable should contain the topic to subscribe to. e.g. events
LISTEN_ADDR environment variable should contain the address to listen on. e.g. localhost
LISTEN_PORT environment variable should contain the port to listen on e.g. 8080
CHECK_SIMILAR=true
ENDIAN=little
FLUSH_MINUTES=5
HIVE_DATABASE=events
HIVE_URL=thrift://localhost:9083
IGNORE_NULLS=true
KAFKA_ERROR_TOPIC=error_firehose
KAFKA_GROUP_ID=filter_2
KAFKA_INPUT_TOPIC=firehose
KAFKA_OUTPUT_TOPIC=firehose_output
KAFKA_POLL_DURATION_MS=5000
KAFKA_SERVERS=localhost:9092
LISTEN_ADDR=0.0.0.0
LISTEN_PORT=3004
ORC_BATCH_SIZE=1000
S3_BUCKET=analytics
S3_PREFIX=test1
SERVICE_KAFKA_TOPIC=firehose
WORKER_KAFKA_TOPIC=firehose_output
DATABASE_URL=jdbc:mysql://localhost:3306/shokuyoku
ERROR_WORKER_KAFKA_TOPIC=error_firehose
DATABASE_USERNAME=admin
DATABASE_PASSWORD=admin
docker run shokuyoku:test worker
You should add environment variables for proper execution.
The worker currently requires an additional environment variable (ORC_SCHEMA
). This environment variable should contain a string-form schema (e.g. struct<mystring:varchar(255),myint:bigint,myarray:array<varchar(255)>,mybool:boolean,mytimestamp:timestamp>
) for the ORC file to be written. This will be replaced with a Hive query in the future.
KAFKA_SERVERS environment variable should contain a comma-separated list of kafka servers. e.g. localhost:9092,localhost:9093
KAFKA_GROUP_ID environment variable should contain the name of the Kafka group. e.g. shokuyoku
KAFKA_TOPIC environment variable should contain the topic to subscribe to. e.g. events
KAFKA_POLL_DURATION_MS environment variable should contain the duration for the Kafka Consumer `poll` method in milliseconds. e.g. 500
FLUSH_MINUTES environment variable should contain the interval between flushes in minutes. e.g. 15
AWS_DEFAULT_REGION environment variable should be set https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-envvars.html
AWS_ACCESS_KEY_ID environment variable should be set https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-envvars.html
AWS_SECRET_ACCESS_KEY environment variable should be set https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-envvars.html
S3_BUCKET environment variable should contain the bucket to write events to. e.g. my-event-bucket
ORC_BATCH_SIZE environment variable should contain the number of records per orc batch. e.g. 1024
The application is currently written using Java 1.8; the goal is to upgrade this to a newer version after considering the upgrade path for the dependencies. Below is an outline of the architecture as well as some norms found in the code base.
App.java
is the main class. This simple class interrogates the command arguments for either service
or worker
, instantiates the approapriate class and starts it.
Service.java
is the primary logic for the listening service. It implements a simple web server using Undertow. This web server currently listens on all paths and expects a JSON body. Once JSON is received this service will write format the JSON according to Firehose format (formats.Firehose
) and send it to the configured Kafka topic (env.KAFKA_TOPIC
).
Worker.java
is the top-level logic for consuming Firehose messages from the Kafka topic. During the consumer loop a map of EventDriver
objects is made for each different event
name pulled from the topic. Each message received from Kafka is sent to the addMessage
method of the appropriate EventDriver
this will add it to the current file. Every env.FLUSH_MINUTES
the consumer will flush
the EventDriver
and commit offsets.
BasicEventDriver.java
is the only implementation of EventDriver
currently available. When initialized this EventDriver
fetchs the schema for the event
and creates an ORC batch. When addMessage
is called it flattens the incoming message using JSONColumnFormat
and then iterates through it adding each value that matches (both name and type) the schema
for the event. If the message added meets the max batch.size
the batch is written to a local file. When flush
is called the local ORC file is closed and pushed to S3.
Environment variables are used for configuration. Environment variables are directly accessed wherever configuration is needed. Both Service
and Worker
validate all required environment variables when initialized using the verifyEnvironment
method.
When a class has a local variable it is directly updated by methods. For example, setColumns
in the BasicEventDriver
class directly updates this.columns
instead of returning an object to be set separately.
Ideally class separation includes interfaces for future expansion. For example BasicEventDriver
implements the EventDriver
interface which can later be used to make a separate interchangeable class that uses a different metastore, file format or storage service.
Whenever possible senseless errors/warnings are fixed. For example main/resources/log4j2.xml
was added to the project with a copy of the default log4j configuration in order to silence log4j complaints.