fluentd Reading json payloads from kafka and insert them to Mariadb

docker-compose.yml

services:
fluentd:
    image: fluent/mariadb-msql:latest
    container_name: fluent-smariadb-msql
    network_mode: host
    volumes:
      - ./fluent.conf:/fluentd/etc/fluent.conf

Dockerfile

FROM fluent/fluentd:v1.16.0-debian-1.0

# Switch to root user for installations
USER root

# Install required tools and dependencies
RUN apt-get update && apt-get install -y \ 
libaio1 build-essential ruby-dev libsnappy-dev default-libmysqlclient-dev && \
apt-get clean && rm -rf /var/lib/apt/lists/*
#Install required gem plugins 
RUN gem install fluent-plugin-kafka --no-document && \
gem install fluent-plugin-sql --no-document && \
gem install db-mariadb && \
gem install mysql2 && \
gem install snappy --no-document && \
gem install fluent-plugin-record-transformer --no-document 

# Copy Fluentd configuration files into the container
COPY fluent.conf /fluentd/etc/
# Set the Fluentd user
USER fluent

fluentd.conf

<source>
  @type kafka
  brokers "localhost:9092"
  topics "person"
  format json

  
</source>



# <match person>
#   <parse>
#   @type json
# </parse>
  # @type stdout
  # format json
# </match>

<match person>
  
  @type sql
  host 127.0.0.1
  port 3306
  database test
  adapter mysql2
  username root
  password qwe123
  socket /tmp/mysql.sock

  <table>
    table "person"
    column_mapping first_name:firstname,last_name:lastname
  </table>
</match>

json payload sample

# generated messages that stored in a kafka topic called person
{
  "firstname": "John",
  "lastname": "Anderson"
}