kafka-connect-s3 : Ingest data from Kafka to Object Stores(s3)
Alternatives To Streamx
Project NameStarsDownloadsRepos Using ThisPackages Using ThisMost Recent CommitTotal ReleasesLatest ReleaseOpen IssuesLicenseLanguage
5 days ago25C#
Full-stack .Net 7 Clean Architecture (Microservices + Dapr, Modular Monolith, Monolith), Blazor, Angular 15, React 18, Vue 3, Domain-Driven Design, CQRS, SOLID, Asp.Net Core Identity Custom Storage, Identity Server, Entity Framework Core, Selenium, SignalR, Hosted Services, Health Checks, Rate Limiting, Cloud (Azure, AWS) Services, ...
Eventhorizon1,4635133 days ago68January 03, 202248apache-2.0Go
Event Sourcing for Go!
Seldon Server1,420
3 years ago44June 28, 201726apache-2.0Java
Machine Learning Platform and Recommendation Engine built on Kubernetes
Devops Bash Tools1,250
a day ago1mitShell
1000+ DevOps Bash Scripts - AWS, GCP, Kubernetes, Docker, CI/CD, APIs, SQL, PostgreSQL, MySQL, Hive, Impala, Kafka, Hadoop, Jenkins, GitHub, GitLab, BitBucket, Azure DevOps, TeamCity, Spotify, MP3, LDAP, Code/Build Linting, pkg mgmt for Linux, Mac, Python, Perl, Ruby, NodeJS, Golang, Advanced dotfiles: .bashrc, .vimrc, .gitconfig, .screenrc, tmux..
Nagios Plugins1,095
23 days ago64otherPython
450+ AWS, Hadoop, Cloud, Kafka, Docker, Elasticsearch, RabbitMQ, Redis, HBase, Solr, Cassandra, ZooKeeper, HDFS, Yarn, Hive, Presto, Drill, Impala, Consul, Spark, Jenkins, Travis CI, Git, MySQL, Linux, DNS, Whois, SSL Certs, Yum Security Updates, Kubernetes, Cloudera etc...
Sitewhere8546192 years ago18June 19, 2017112otherJava
SiteWhere is an industrial strength open-source application enablement platform for the Internet of Things (IoT). It provides a multi-tenant microservice-based infrastructure that includes device/asset management, data ingestion, big-data storage, and integration through a modern, scalable architecture. SiteWhere provides REST APIs for all system functionality. SiteWhere provides SDKs for many common device platforms including Android, iOS, Arduino, and any Java-capable platform such as Raspberry Pi rapidly accelerating the speed of innovation.
15 days ago21mitTypeScript
📊 📊 📊 Monitors the health and web traffic of servers, microservices, Kubernetes/Kafka clusters, containers, and AWS services with real-time data monitoring and receive automated notifications over Slack or email.
2 months ago7mitJupyter Notebook
Code for Agile Data Science 2.0, O'Reilly 2017, Second Edition
Kafka Connect Storage Cloud239
18 hours ago156otherJava
Kafka Connect suite of connectors for Cloud storage (Amazon S3)
3 years ago8May 08, 201822apache-2.0Go
Serverless Platform for the stateful services
Alternatives To Streamx
Select To Compare

Alternative Project Comparisons

THIS REPO IS ARCHIVED Based on Security issues SEC ticket #SEC-2988

StreamX: Kafka Connect for S3

Forked from the awesome kafka-connect-hdfs

StreamX is a kafka-connect based connector to copy data from Kafka to Object Stores like Amazon s3, Google Cloud Storage and Azure Blob Store. It focusses on reliable and scalable data copying. It can write the data out in different formats (like parquet, so that it can readily be used by analytical tools) and also in different partitioning requirements.

##Features :

StreamX inherits rich set of features from kafka-connect-hdfs.

  • Support for writing data in Avro and Parquet formats.
  • Provides Hive Integration where the connector creates partitioned hive table and periodically does add partitions once it writes a new partition to s3
  • Pluggable partitioner :
  • default partitioner : Each Kafka partition will have its data copied under a partition specific directory
  • time based partitioner : Ability to write data on hourly basis
  • field based partitioner : Ability to use a field in the record as custom partitioner

In addition to these, we have made changes to the following to make it work efficiently with s3

  • Exactly-once guarantee using WAL
  • Support for storing Hive tables in Qubole's hive metastore (coming soon)

##Getting Started:

Pre-req : StreamX is based on Kafka Connect framework, which is part of Kafka project. Kafka Connect is added in Kafka 0.9, hence StreamX can only be used with Kafka version >= 0.9. To download Kafka binaries, check here.

Clone : git clone

Branch : For Kafka 0.9, use 2.x branch. For Kafka 0.10 and above, use master branch.

Build : mvn -DskipTests package

Once the build succeeds, StreamX packages all required jars under target/streamx-0.1.0-SNAPSHOT-development/share/java/streamx/* in StreamX repo. This directory needs to be in classpath.

Add Connector to Kafka Connect Classpath:

export CLASSPATH=$CLASSPATH:`pwd`/target/streamx-0.1.0-SNAPSHOT-development/share/java/streamx/*

Start Kafka Connect

In Kafka, change the following in config/ or config/ depending on what mode you want to use.

bootstrap.servers=set Kafka end-point (ex: localhost:9092)

Use ByteArrayConverter to copy data from Kafka as-is without any changes. (copy JSON/CSV)

Run Kafka Connect in Standalone mode

Set s3.url and hadoop-conf in StreamX config/ StreamX packages hadoop-conf directory at config/hadoop-conf for ease-of-use. Set s3 access and secret keys in config/hadoop-conf/hdfs-site.xml.

In Kafka, run

bin/connect-standalone etc/kafka/ \

You are done. Check s3 for ingested data!

Run Kafka Connect in distributed mode
bin/ config/

We have started the Kafka Connect framework and the S3 Connector is added to classpath. Kafka Connect framework starts a REST server (rest.port property in listening for Connect Job requests. The copy job can be submitted by hitting the REST end-point using curl or any REST clients.

For example, to submit a copy job from Kafka to S3

curl -i -X POST \
   -H "Accept:application/json" \
   -H "Content-Type:application/json" \
   -d \
}}' \
  • Uses S3SinkConnector
  • Uses SourceFormat, which copies the data as-is (Note that this needs to be used with ByteArrayConverter)
  • tasks.max refers to number of tasks that copies the data
  • a new file is written after flush.size number of messages
  • S3 Configuration It uses the hadoop file system implementation (s3a/s3n) to write to s3. The connect job has a configuration called hadoop.conf.dir and this needs the directory where hdfs-site.xml and other hadoop configuration resides. StreamX packages the hadoop dependencies, so it need not have hadoop project/jars in its classpath. So, create a directory containing hadoop config files like core-site.xml, hdfs-site.xml and provide the location of this directory in hadoop.conf.dir while submitting copy job. (StreamX provides a default hadoop-conf directory under config/hadoop-conf. Set your s3 access key, secret key there and provide full path in hadoop.conf.dir)

You have submitted the job, check S3 for output files. For the above copy job, it will create s3://streamx/demo/topics/adclicks/partition=x/

Note that, a single copy job could consume from multiple topics and writes to topic specific directory.

To delete a Connect job,

curl -i -X DELETE \
   -H "Accept:application/json" \
   -H "Content-Type:application/json" \

To list all Connect jobs,

curl -i -X GET \
   -H "Accept:application/json" \
   -H "Content-Type:application/json" \

Restarting Connect jobs : All Connect jobs are stored in a Kafka Queue. So, restarting the Kafka Connect will restart all the connectors submitted to it.

Docker Streamx supports Docker, but only in distributed mode To build your image,

docker build -t qubole/streamx .

When you run your container, you can override all the properties in file with env vars. env_vars will be of format CONNECT_BOOTSTRAP_SERVERS corresponding to bootstrap.servers. The convention is to prefix env vars with CONNECT. Example of how to run your container,

docker run -d -p 8083:8083 --env CONNECT_BOOTSTRAP_SERVERS=public_dns:9092 --env CONNECT_AWS_ACCESS_KEY=youracesskey --env CONNECT_AWS_SECRET_KEY=yoursecretkey qubole/streamx

You can also use Avro/Parquet format. Example:

docker run -d -p 8083:8083 --env CONNECT_BOOTSTRAP_SERVERS=public_dns:9092 --env CONNECT_AWS_ACCESS_KEY=youracesskey --env CONNECT_AWS_SECRET_KEY=yoursecretkey  --env CONNECT_KEY_CONVERTER=io.confluent.connect.avro.AvroConverter --env CONNECT_VALUE_CONVERTER=io.confluent.connect.avro.AvroConverter --env CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL= --env CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL= qubole/streamx


  • Support other object stores like Google Cloud Storage and Azure Blob Store
  • Currently, data can be written in avro/parquet format. This project will add support for more formats
  • Deal with features related to s3, like small-file consolidation
Popular Kafka Projects
Popular Amazon Web Services Projects
Popular Data Processing Categories
Related Searches

Get A Weekly Email With Trending Projects For These Categories
No Spam. Unsubscribe easily at any time.
Big Data
Kafka Connect