What connects an IT company from Sarajevo with one of the largest shipping firms in the U.S.? The answer is their shared vision for innovation.

This story is about building a successful long-term collaboration with our client and crafting several products that helped them overcome some of their main challenges — scaling their business and bringing several cost reductions in an industry that was once resistant to change.

The Rise of Digitalization in the Shipping Industry

With the constant growth of technology and the countless benefits digitalization has brought to various areas of life, the shipping industry has also opened its doors to automation. And our client recognized this from the very beginning. Investing in digitalization has led them to be one of the fastest-growing shipping companies in the US.

Seizing Opportunity in a Time of Crisis

At the beginning of 2020, a difficult time of the COVID-19 pandemic, when the world faced uncertainty, so did the IT industry. But for us, an opportunity for collaboration emerged, opening up a new path during a critical time. Our client recognized a need for a product that would automate and enhance one of its processes. What initially began as developing what was a primary task; as trust was established and quality was achieved, new project opportunities began to arise. Thus, in Atlantbh, a small but dedicated team of six people was formed, ready to take on this challenge.

Marking the Starting Line

You’ve likely purchased a frozen product at some point. When buying such products, they are kept at precisely controlled temperatures in the store, and we’ve often heard warnings against refreezing once-thawed items. However, the journey of these initially frozen products to markets across the country is a process most consumers rarely think about. Frozen and temperature-sensitive goods are transported in refrigerated trucks with strictly controlled temperatures. Even minor fluctuations can lead to spoilage and significant losses.

Before collaborating with us, the systems in place to monitor these highly sensitive temperature changes in refrigerated trucks were decentralized and required human intervention to address any issues. Therefore, it wasn’t a scalable system and had possibilities for failure as it heavily relied on the people operating the system. Automation of this system was a natural step forward, and that’s where we fit in with our first project — Trailer Monitoring.

The team that started work on the project included three Software Engineers, one Quality Assurance Engineer, one DevOps Engineer, a UX/UI designer, and a Product Manager on the team as well.

From Concept to Code

The project started with the goal of having a working production solution in two months. To achieve this, we worked on delivering a backend-only Spring Boot project with an event-driven architecture consisting of event generators, event processors, and a notification engine.

This approach allowed us to have a seamless data flow from receiving an external update to sending a notification email. Decoupling event generation, processing, and notifying allowed us to have a scalable, maintainable design.

• Event generators were isolated allowing for independent data processing in parallel.
• Event processors handled all of the business logic, and they were clearly separated from the rest of the system.
• The notification engine had a simple entry point allowing us to easily add or change the communication method in the future.

No external libraries were used for sending and receiving these events, but rather a simple producer > consumer mechanism, built to allow scaling and future changes.

We knew upfront that the number of events could be large and that we had to store each event in order to know exactly what happened at any moment. As a point of reference, a trailer sent an update approximately every hour, and there were hundreds of trailers. After researching ways to keep up with this volume of data, we ended up using TimescaleDB with PostgreSQL. This proved to be a superb solution allowing for storing and querying millions of rows in a matter of milliseconds. We liked it so much that we even wrote another about it.

From the pool of millions of rows of data, there was the need to make geospatial computations (distance calculations, closest points, etc.) To support this, we extended PostgreSQL with and also stored the latest state for each trailer. This made querying the current trailer state, all the trailers at some location, or the trailer path easy and fast, even after the trailer fleet expanded.

The project quickly expanded with the emergence of requests for a user interface. The UI started as a reporting dashboard at first, but over time, it became an essential tool for monitoring and controlling the trailer fleet, and it was included in the client’s onboarding process.

• The maintenance crew used it to check for faulty trailers.
• Operation managers used it to check the daily audit and statistics.
• Driver managers used it to always be up to date on the trailer state.
• Admins for controlling notifications in real-time and setting schedules so no emails get lost.

Requests for new features were pouring in, and the trust we had established pushed us into new waters as a new project was around the corner.

Loading Things Up

Thanks to the success and value of our collaboration, we were given the opportunity to expand our team and work in another important branch of the client’s business, — dealing with finding the right carrier to outsource transportation of certain goods (referenced as load in the text that follows). At the time, this whole process was completely manual — making inquiries, agreeing on the price for each load and handling common scenarios such as last-minute cancellations. The importance of this process and the scale at which it was performed made its automation a natural next step. A new team was formed, which ended up sharing members across different teams. This setup posed challenges as it introduced a new team dynamic, but it also had advantages — business logic often overlapped, and workload distribution across projects allowed for more flexibility.

The product aimed to automate the whole process from a load becoming available for transport, through offering it to specific carriers, to finally putting it up for auction where any partnering carrier could bid for it. With the new domain also came new challenges and considerations:

We explored multiple user management services, finally settling on using . We aimed to create a user-friendly solution that represents the client’s company in the best way, and that would scale with the increased number of users. We had to put procedures in place for onboarding users to a new tool and working on the product while it was actively used in production.

In case a load wasn’t accepted by any carrier it was offered to, it would get published to an auction where any of the almost a thousand partners could bid for it using the application. It was paramount to ensure the correct state of the system after a possibly large amount of user interactions at the same time. To prevent race conditions, we explored different locking solutions . Testing this feature was a challenge in itself, and our QA team kept improving the automation suite over time, challenging the limits of the application and exploring complex scenarios.

Besides the carrier-facing UI that lets them keep track of offers and ongoing auctions, we implemented a dashboard aimed at brokers — users monitoring the complete lifecycle of a load. It surfaced a myriad of actions they could take to resolve even the most complicated edge cases that would ultimately come up. This dashboard utilized a based communication protocol so the data presented on the dashboards would update in real-time.

The application was deployed to production after a year of development, and it has since been an invaluable resource in automating a high-yield business process.

The Golden Age

We were actively working on a couple of projects involving over 20 team members. The systems grew, as did our client’s trust and satisfaction. In 2023, a new opportunity emerged. Our client recognized the need to automate one of their HR processes. After extensive discussions and work on the MVP, we realized that the scope of the work extended far beyond improving a single process. With valuable experience behind us, this was an opportunity to start a project and adopt all the lessons we learned.

This system encompassed monitoring one of their yards, tracking personnel and equipment moving in and out of it, conducting various employee performance surveys, generating diverse analytics and behavioral insights, and even an in-app domain-specific task management system.

One aspect we are particularly proud of-and which significantly streamlined the development of this product-was our design process and established design system. Our design processes evolved with our projects over the years. We started with one UX/UI designer and ended with a team of four designers. Over the years, they have been building a reusable design system consisting of custom components in , which reached its peak at the time we started working on this project. Custom components were translated into UI components and used as building blocks for complex pages. This sophisticated design system, complete with a , helped turn frontend development into an intuitive, puzzle-like experience.

A distinctive characteristic of this project is that a lot of user input came from complex multi-stage questionnaires.

• This required a scalable data model that supported different types of questions (text, multiple choice, single choice, answering by choosing from a map, etc.) as well as many ways to group them to create very specific forms
• Forms on the UI were fully dynamic, powered by the backend service that was responsible for storing the full structure of the questionnaire, specific questions, and their answers
• Later improvements included autosave capabilities and mechanisms for answering questions through integration with external services

Lasting Impact: More Than Just Software

Back in 2020, we didn’t know much about the complexity of the shipping industry and its core businesses. It is actually a lot more than moving different kinds of goods from point A to point B. Its role is crucial in global supply chains, ensuring products reach businesses and consumers efficiently. The systems we built covered key aspects of this industry: fleet management, logistics coordination, and regulatory compliance, all of which contribute to efficient and timely delivery.

Four years later, hard work and dedication have exceeded expectations. The team has grown significantly and set up four different tracks, developing products of great strategic value to the client, improving their key processes, and leaving a lasting impact on the entire organization. Many business processes were automated and centralized, leading to significant cost reductions and general business scalability. As the projects grew, so did our team and individuals. We learned a lot and established some of the procedures that are used on other projects as well. Over this period, more than 30 individuals contributed to these projects, building a shared legacy that the Atlantbh company will rely on in the future.

Originally published at https://www.atlantbh.com on March 4, 2025.

When Shipping Meets Code: A Story of Collaboration and Innovation was originally published in Atlantbh Engineering on Medium, where people are continuing the conversation by highlighting and responding to this story.

In the previous blog we explained what Kafka Connect is, what we can do with it, and what are some of the most important components that we can use to establish a reliable and fault-tolerant big data pipeline.

In this blog, we will deploy a small-scale but production-ready Kafka Connect cluster where we will take Avro and Protobuf messages from different topics and store them in Snowflake tables. We will use Schema Registry to validate our Avro and Protobuf schemas.

In theory, it seems like a straightforward process but in practice, it can get a bit complex especially if you are setting up Kafka Connect for the first time. So many properties can be easily overlooked, but precisely those properties can save you from losing your precious Kafka messages.

We will go over the whole process and show you how we can set up a production-ready pipeline from scratch.

Environment setup

• Kubernetes cluster*
• Kafka cluster
• Schema registry
• Snowflake account
  *Kubernetes cluster is not required, but if you plan to set up Kafka Connect somewhere else (like EC2, VMs), you must make some adjustments.

All the above services (besides Snowflake) can be deployed locally or you can use managed services.
To make this blog less complex since the main focus is on Kafka Connect, (for our example), we will use Confluent Platform for the Kafka cluster and schema registry.

We understand that not everybody can use Confluent Platform for commercial projects, and you do not have to. The process is not tied to Confluent, which can be achieved without using any Confluent products; instead, it can be done using open-source software.For reference, our company is not using anything related to Confluent, not even the schema registry or converters.
If you want to use Confluent Platform to follow along, there is a 30 day free trial that is more than enough to replicate everything from this blog. The same 30 day trial exists in Snowflake too.

To confirm that it also works without the Confluent Platform we also deployed Kafka using Koperator on our Kubernetes cluster, and the setup was the same. For the schema registry, you can try out Karapace.

Setting up Kafka Connect to store messages in Snowflake

To make our example more straightforward, we will imagine that we are a sushi restaurant that uses Kafka, Kubernetes and Snowflake to manage their customers, orders and reviews (this is probably not the best approach but is easier to understand). Even though we will use a small sample dataset, the same setup will work with any dataset size.

Our company is using a similar pipeline in production to process millions of records each day, and it works without any issues. Kafka Connect clusters are horizontally scalable, which means that you will only need to increase the replica count to support larger workloads.

Our application will have three topics: two will use Protobuf, and one will use Avro. We wanted to cover multiple data serialization formats to demonstrate how to handle various different Kafka Connect clusters.

The diagram that we will follow to set up our Kafka to Snowflake pipeline is below:

We will be using Kubernetes, but if you want to deploy it somewhere else, the Kafka Connect setup is the same; the only difference is how you want to deploy it.

Feel free to use this Docker image dksadx/kafka-connect-snowflake, or if you want to deploy it on bare metal, you can use this Dockerfile as a guide.

Setting up the demo environment (optional)

This section is optional; if you already have Kafka, Kubernetes, and Snowflake set up, feel free to skip this and go straight to the Configuring Kafka Connect section.

If you want to follow along, we need to set up a demo environment first.

Creating a Kubernetes cluster

We will use Podman with Kind to set up a local Kubernetes cluster. You can use any other tool to set up your cluster or deploy a managed cluster via your preferred cloud provider.

# Install Podman, QEMO, Kind and kubectl (MacOS only)
$ brew install podman qemu kind kubectl helm

# Create and start a VM for the cluster (increase resources if deploying Kafka)
$ podman machine init - cpus 4 - disk-size 30 - memory 8192 - now demo-vm

# Create a Kubernetes cluster with Kind
$ KIND_EXPERIMENTAL_PROVIDER=podman kind create cluster - name kafka_connect_demo

# Set the context
$ kubectl config set-context kafka_connect_demo

Check if everything works correctly:

$ kubectl get all
#   NAME                 TYPE        CLUSTER-IP   EXTERNAL-IP   PORT(S)   AGE
#   service/kubernetes   ClusterIP   10.96.0.1    <none>        443/TCP   2m48s

Setting up Kafka on Kubernetes

If you do not have a Kafka cluster, you can create one for free via the Confluent Platform.
After creating the cluster, create three topics called avro-customers, proto-orders, and proto-reviews.

If you do not want to use the Confluent Platform and instead want to deploy it locally on the Kubernetes cluster we previously created, the easiest way to set it up is via the Koperator Helm chart. To set up the latest and most up-to-date version of the chart, use their guide, which can be found in the README.md here.

After the setup, we can create three topics for our sushi demo app.

kubectl create -n kafka -f - <<EOF
apiVersion: kafka.banzaicloud.io/v1alpha1
kind: KafkaTopic
metadata:
    name: avro-customers
spec:
    clusterRef:
        name: kafka
    name: customers
    partitions: 1
    replicationFactor: 1
    config:
        "retention.ms": "604800000"
        "cleanup.policy": "delete"
EOF

Do the same for the other two topics: proto-reviews, and proto-orders.

Setting up the Snowflake database

For storing Kafka messages, we will use Snowflake, and thus, we will need a Snowflake account. If you do not have an account, you can sign up for a free 30-day trial on their page: www.snowflake.com.
After creating an account, we will need to create a warehouse. If you are not familiar with Snowflake, warehouses are basically compute instances in Snowflake.

You can create a warehouse by going to the Admin panel and clicking the blue + WAREHOUSE button in the top right corner. Choose XS size for this demo.

After creating a warehouse, go to Projects -> Worksheets -> SQL Worksheet.

Now, we need to configure Snowflake so that we can actually use the Kafka Connect connector.

There are few steps that we have to do:

1. Create a database and schema:

CREATE DATABASE kafka_connect_demo;
CREATE SCHEMA kafka_connect_demo.sushi_restaurant;

2. Create user, role and configure all the privileges:

-- Use a role that can create and manage roles and privileges.
USE ROLE accountadmin;

-- Create a Snowflake role with the privileges to work with the connector.
CREATE ROLE kafka_connector_role_sushi;

-- Create a Snowflake user that will be used by the kafka connect connector to access Snowflake.
CREATE USER IF NOT EXISTS kafka_connect_sushi_user
    PASSWORD = "my_very_strong_password"
    LOGIN_NAME = "kafka_connect_sushi_user"
    DEFAULT_ROLE = "kafka_connector_role_sushi"
    DEFAULT_SECONDARY_ROLES = ('ALL');

-- Grant privileges on the database.
GRANT USAGE ON DATABASE kafka_connect_demo TO ROLE kafka_connector_role_sushi;

-- Grant privileges on the schema.
GRANT USAGE ON SCHEMA kafka_connect_demo.sushi_restaurant TO ROLE kafka_connector_role_sushi;
GRANT CREATE TABLE ON SCHEMA kafka_connect_demo.sushi_restaurant TO ROLE kafka_connector_role_sushi;
GRANT CREATE STAGE ON SCHEMA kafka_connect_demo.sushi_restaurant TO ROLE kafka_connector_role_sushi;
GRANT CREATE PIPE ON SCHEMA kafka_connect_demo.sushi_restaurant TO ROLE kafka_connector_role_sushi;

-- Grant the custom role to an existing user.
GRANT ROLE kafka_connector_role_sushi TO USER kafka_connect_sushi_user;

-- Grant the custom role to your user (assuming you are using ACCOUNTADMIN)
GRANT ROLE kafka_connector_role_sushi TO USER ACCOUNTADMIN;

-- Set the custom role as the default role for the user.
-- If you encounter an 'Insufficient privileges' error, verify the role that has the OWNERSHIP privilege on the user.
ALTER USER kafka_connect_sushi_user SET DEFAULT_ROLE = kafka_connector_role_sushi;

-- Grant privileges on a warehouse to our role.
GRANT ALL PRIVILEGES ON WAREHOUSE kafkaconnect TO ROLE kafka_connector_role_sushi;

3. Generate an RSA key pair. If you are unfamiliar with how to do that, you can use this guide: Using key pair authentication & key rotation.

4. Add the public key to the user you created in step 2, in our case the user is called kafka_connect_sushi_user.
You must remove all the new lines and also the headers and footers ( — — -BEGIN PUBLIC KEY — — — , — — -END PUBLIC KEY — — -) from the public key.

-- Add RSA public key to user. This is how your public key needs to look like, without headers and new lines.
ALTER USER kafka_connect_sushi_user SET RSA_PUBLIC_KEY='MIIBIj...';

5. While still signed in to Snowflake, go to the home pages and save the url. The url should look like this: https://app.snowflake.com/<orgname>/<accountname>/worksheets.
We will need this information when setting up Kafka Connect.

Setting up Schema Registry

Having a schema registry is not required, and you can do the whole setup without it but I highly recommend using it. If you are working or planning to work on a larger project, having a schema registry can save you a lot of trouble. If you do not plan to use schema registry, skip this section and remove all config properties that have schema in its name and you should be good to go.

To set up a schema registry there are two choices:

1. sign up to Confluent Platform and use schema registry as a service,
2. host your own.

Whichever option you choose, the rest will be the same; only the self-hosted one will require additional steps. To reduce the complexity of the blog, we will use Confluent Platform for this demo.

If you choose to use Confluent Platform, go to their site, create an account, and start a 30-day free trial.

After finishing all the required steps, your environment and cluster should be ready. The only thing that we need to do here is find the endpoint url and generate an API key. You can do that by clicking on your environment and in the right sidebar you can find the endpoint. Below the endpoint, you can generate an API key. Save the above info because we will need it in the next step.

Configuring Kafka Connect

After everything is set up and deployed, we are ready for the main part, configuring and deploying Kafka Connect clusters.
For our example, we will need three clusters since we have avro, protobuf, and string converters.

We will configure and deploy Kafka Connect in a distributed mode because it is the most reliable and fault-tolerant. In most cases, you will want to use this mode.

There are two configuration files that we need to edit:

1. connect-distributed.properties — a file that configures the Kafka Connect cluster,
2. snowflake.json — file that configures the Snowflake connector.

I will not go into too much detail about what all properties mean because you can find that in the official documentation. I will go over the most interesting ones for our use case.

connect-distributed.properties
This file configures the Kafka Connect cluster, it does not configure any connectors. We will add connectors after the cluster is running and in a healthy state.

Since we are deploying our cluster in distributed mode, we will need three additional topics that will be used to store the configs, statuses, and offsets for the clusters.

This is what the configuration for our avro cluster looks like:

# Unique name for the cluster, used in forming the Connect cluster group. Note that this must not conflict with consumer group IDs
group.id=kafka-connect-avro

# Topic to use for storing offsets. This topic should have many partitions and be replicated and compacted.
# Kafka Connect will attempt to create the topic automatically when needed, but you can always manually create
# the topic before starting Kafka Connect if a specific topic configuration is needed.
offset.storage.topic=kafka-connect-avro-offsets
offset.storage.replication.factor=3
offset.storage.partitions=3

# Topic to use for storing connector and task configurations; note that this should be a single partition, highly replicated,
# and compacted topic. Kafka Connect will attempt to create the topic automatically when needed, but you can always manually create
# the topic before starting Kafka Connect if a specific topic configuration is needed.
config.storage.topic=kafka-connect-avro-configs
config.storage.replication.factor=3
config.storage.partitions=1

# Topic to use for storing statuses. This topic can have multiple partitions and should be replicated and compacted.
# Kafka Connect will attempt to create the topic automatically when needed, but you can always manually create
# the topic before starting Kafka Connect if a specific topic configuration is needed.
status.storage.topic=kafka-connect-avro-statuses
status.storage.replication.factor=3
status.storage.partitions=3

You can create the three above topics yourself or let Kafka Connect create it for you.

To configure the schema registry, we will use the information that we got from the previous step:

value.converter.basic.auth.credentials.source=USER_INFO
value.converter.basic.auth.user.info=YOUR_API_KEY:YOUR_API_SECRET
value.converter.expected.schema.name=customers
value.converter.schema.registry.url=https://YOUR_ACCOUNT.confluent.cloud

The most interesting properties for us are the converter properties because if those are wrong, we will not be able to serialize or deserialize messages correctly.

We need to define which converters we want to use for both keys and values. Usually you can use StringConverter for keys but for values you need specific converters.

There are many converters out there, some are officially supported and some are custom made for specific use cases by the community. An example of such a converter is registryless-avro-converter, which is an avro converter that does not depend on the Confluent Schema Registry.

We will use 3 types of converters in our example:

1. Avro — io.confluent.connect.avro.AvroConverter
2. Protobuf — io.confluent.connect.protobuf.ProtobufConverter
3. String — org.apache.kafka.connect.storage.StringConverter

key.converter=org.apache.kafka.connect.storage.StringConverter
# Using avro converter
value.converter=io.confluent.connect.avro.AvroConverter

snowflake.json
This file configures the Snowflake connector for Kafka Connect. It is a JSON file with all the secrets and configurations required for the connector to function. Since this is a connector configuration file, we do not specify it while starting the cluster; instead, we need to send a POST request with the file content as payload to the Kafka Connect endpoint once the cluster is up and running.

You can send the request from anywhere; it does not need to be from the container or pod, but I recommend that you do not expose the Kafka Connect endpoint externally if you really do not need to, mainly for security reasons.

Having the snowflake.json file along with the connect-distributed.properties file can be beneficial, as it allows us to recreate the connector if something unexpected happens automatically.
If you are running the cluster for months, something may break the cluster, and having a backup solution will be a lifesaver in those moments.

It rarely happens, but occurs sometimes, so configuring liveness and startup probes to take care of this is a big plus.

You can find detailed instructions on configuring a Snowflake connector in their official documentation here.
Most of the configuration is self-explanatory but I will point out a few fields that may cause problems:

name
Name of the connector. This name will be part of the url, and it will be used when performing any actions on the connector, e.g., updating the config properties. If you plan to have multiple connectors on a single cluster, make sure they have different names.

snowflake.url.name
Format for the url is <orgname>-<accountname>.snowflakecomputing.com:443. You can find the <orgname> and <accountname> in the url when you log in to your account via the browser.
The url should be similar to this: https://app.snowflake.com/kalsvgw/ab20210/worksheets where <orgname> is kalsvgw and <accountname> is ab20210.

snowflake.private.key
This is the encrypted RSA key that we generated in the section where we did the Snowflake setup. Make sure that you remove all line breaks, whitespace and headers/footers before pasting the key.

snowflake.private.key.passphrase
This is the passphrase used while creating the key.

Deploying Kafka Connect clusters

We are almost ready to deploy our Kafka Connect clusters once the configurations for Kafka Connect and Snowflake connector are ready.

Starting the Kafka Connect in distributed mode is as simple as running the connect-distributed.sh script that comes with every Kafka installation and providing the configuration properties as the first argument:

$ ./connect-distributed.sh connect-distributed.properties

Keep in mind that you need different dependencies (jar files) for different connectors and converters. To see which dependencies are required for our example you can check this Dockerfile.

If everything is configured correctly, we will see that three new topics for offsets, statuses, and configs appeared in our Kafka cluster. These topics are used by the cluster instances/replicas to communicate with each other and distribute the load.

After the cluster is up and there are no errors in the logs, we can add connectors. To add a new connector, we need to send a POST request to the cluster’s endpoint:

curl -X POST -H "Content-Type: application/json" - data @snowflake.json http://localhost:8083/connectors

We should see messages arriving in our Snowflake tables if everything is configured correctly.

Since we will be deploying the cluster in Kubernetes for our sushi restaurant example, we prepared a Helm chart that will take care of most of the setup. We only have to configure some of the cluster and connector properties like bootstrap servers and credentials.

1. First we need to get the chart

git clone https://github.com/DKSadx/kafka-connect-blog.git
cd deploy/helm

2. You will see three types of configuration files in the configs directory, connect-distributed-<TYPE>.properties, snowflake-<TYPE>.json, and config.yaml. Fill out the empty fields with your Kafka and Snowflake information in all of those configs.
If you are confused about what config.yaml is, that is a configuration file that is used by our simple demo Go app called sushi-restaurant-kafka-demo-app.
The purpose of the app is to read some example dummy messages from app/data/ directory and publish those messages to corresponding topics specified in config.yaml.
Besides topics, the configuration file also contains information about the Kafka cluster so that it can connect and publish those messages.

3. After everything is configured, we are ready to install the helm chart:

helm install sushi-restaurant .

Let’s check the pods and wait for everything to be in RUNNING state:

$ kubectl get pods
NAME                                                          READY   STATUS      RESTARTS   AGE
demo-app-dqw6n                                                0/1     Completed   0          9s
sushi-restaurant-customers-avro-0-c8d7c8689-brmm7             1/1     Running     0          9s
sushi-restaurant-customers-avro-0-c8d7c8689-sh4h9             1/1     Running     0          9s
sushi-restaurant-dlq-string-2-56f7cf687-58qb9                 1/1     Running     0          9s
sushi-restaurant-dlq-string-2-56f7cf687-kwhpg                 1/1     Running     0          9s
sushi-restaurant-reviews-orders-protobuf-1-6546fd87fb-5882s   0/1     Running     0          9s
sushi-restaurant-reviews-orders-protobuf-1-6546fd87fb-9dskd   0/1     Running     0          9s

When you inspect the logs of the demo app, they need to look like this. If that is the case, that means that the messages are being published to our topics:

$ kubectl get logs demo-app-dqw6n

2024/09/02 12:08:40 Created Producer rdkafka#producer-1
2024/09/02 12:08:40 Publishing orders to proto-orders topic
2024/09/02 12:08:40 Publishing customers to avro-customers topic
2024/09/02 12:08:40 Publishing reviews to proto-reviews topic
2024/09/02 12:08:41 Delivered message to topic proto-reviews [1] at offset 0
2024/09/02 12:08:41 Delivered message to topic proto-orders [1] at offset 0
2024/09/02 12:08:41 Delivered message to topic avro-customers [0] at offset 0
2024/09/02 12:08:41 Delivered message to topic proto-orders [1] at offset 1
2024/09/02 12:08:41 Delivered message to topic avro-customers [1] at offset 0
2024/09/02 12:08:41 Delivered message to topic proto-orders [1] at offset 2

When you now check logs of any of the Kafka Connect pods, if the log looks similar to this, then it means that messages are successfully being consumed from topics and pushed to Snowflake:

$ kubectl logs sushi-restaurant-customers-avro-0-c8d7c8689-6gqkj 

2024-09-02 12:11:32,784] INFO [snowflake_sushi_customers|task-2] [SF_INGEST] Channel=CUSTOMERS_2 created for table=CUSTOMERS (net.snowflake.ingest.streaming.internal.SnowflakeStreamingIngestChannelInternal:58)
[2024-09-02 12:11:33,140] INFO [snowflake_sushi_customers|task-0] [SF_KAFKA_CONNECTOR] Fetched offsetToken for channelName:KAFKA_CONNECT_DEMO.SUSHI_RESTAURANT.CUSTOMERS.CUSTOMERS_0, offset:1 (com.snowflake.kafka.connector.internal.streaming.TopicPartitionChannel:46)
[2024-09-02 12:11:33,168] INFO [snowflake_sushi_customers|task-0] [SF_KAFKA_CONNECTOR] task opened with 1 partitions, execution time: 2955 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask:46)
[2024-09-02 12:11:33,169] INFO [snowflake_sushi_customers|task-0] [Consumer clientId=connector-consumer-snowflake_sushi_customers-0, groupId=avro-customers] Seeking to offset 2 for partition customers-0 (org.apache.kafka.clients.consumer.KafkaConsumer:1585)
[2024-09-02 12:11:33,469] INFO [snowflake_sushi_customers|task-2] [SF_KAFKA_CONNECTOR] Fetched offsetToken for channelName:KAFKA_CONNECT_DEMO.SUSHI_RESTAURANT.CUSTOMERS.CUSTOMERS_2, offset:2 (com.snowflake.kafka.connector.internal.streaming.TopicPartitionChannel:46)
[2024-09-02 12:11:33,471] INFO [snowflake_sushi_customers|task-2] [SF_KAFKA_CONNECTOR] task opened with 1 partitions, execution time: 3255 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask:46)
[2024-09-02 12:11:33,471] INFO [snowflake_sushi_customers|task-2] [Consumer clientId=connector-consumer-snowflake_sushi_customers-2, groupId=avro-customers] Seeking to offset 3 for partition customers-2 (org.apache.kafka.clients.consumer.KafkaConsumer:1585)
[2024-09-02 12:11:36,038] INFO [snowflake_sushi_customers|task-0] [SF_KAFKA_CONNECTOR] Fetched offsetToken for channelName:KAFKA_CONNECT_DEMO.SUSHI_RESTAURANT.CUSTOMERS.CUSTOMERS_0, offset:1 (com.snowflake.kafka.connector.internal.streaming.TopicPartitionChannel:46)

You are probably wondering, how does it work if we did not add the Snowflake connector. The reason for this is that we automated the process using Startup Probes and this script. The script will check if the connector exists, and if not, it will add it automatically.

Now, if we go to Snowflake and check any of the new tables, you will see the example messages there:

Dead letter queue to the rescue

Great, the pipeline works, but what happens if somebody sends a message that was serialized using a different converter, e.g., sends a protobuf message on an avro topic.

We can not ignore those messages, maybe those are some really important transactions that were sent by mistake.

To handle such cases, we need a dead letter queue or DLQ.

If you remember from the previous blog or check the example configuration file snowflake.json in the Helm chart, you will see that DLQ properties are already configured:

"errors.deadletterqueue.topic.name": "kafka_connect_dlq",
"errors.deadletterqueue.topic.replication.factor": 3,
"errors.deadletterqueue.context.headers.enable": true

You will also see one more property that is not directly related to DLQ but is important to set it up correctly.

"errors.tolerance": "all"

By default, error.tolerance is set to none, which means that if a message can not be deserialized, the Kafka Connect instance will throw an error and stop.

For some use cases, that can be exactly what we want, but for some, we can not afford to stop the whole pipeline if someone publishes a message to a wrong topic. In such cases it is good to have a Kafka Connect cluster specifically for DLQ with the property error.tolerance set to all.

With that approach you have a dedicated cluster that will take all the failed messages that went to the DLQ and push those messages to Snowflake as they are.

Now, we can continue consuming messages while also saving the failed messages to be analyzed later. For production environments, you will want to configure alerts on the Snowflake side so that you will be notified if new messages are added to the DLQ table.

Conclusion

When faced with the challenge of transferring messages from Kafka to another destination, in our case, it was Snowflake, Kafka Connect will be the best approach for most cases.

We used Snowflake as an example since we are using it heavily on our projects and it is becoming a go-to platform when it comes to big data pipelines. Snowflake is just one sink connector, but there are hundreds more. If you want to take messages from Kafka topics and upload them somewhere or vice-versa, there is a good chance that someone already wrote a connector for that.

Having written custom consumers and discovering bugs when least expecting them is never a good feeling. That is one of the reasons why using a battle-tested tool with many users and a large community behind it is, in most cases, the way to go.

Consuming and Storing Kafka Messages in Snowflake Using Kafka Connect (Part 2) was originally published in Atlantbh Engineering on Medium, where people are continuing the conversation by highlighting and responding to this story.

Every year, zettabytes of data are transferred over the Internet. Managing, processing, and storing the data can be a really complex task and requires cutting-edge tools. Many tools can be used, but two technologies that are almost essential to every modern data pipeline: Kafka and Snowflake.

If you are planning to stream a large amount of data, Kafka is usually the best choice. Scalability, reliability, low latency and popularity are reasons why Kafka stands out. Kafka is by far the most popular event streaming platform and a go-to for many cases when it comes to streaming data.

Since you need to process and store the data somewhere, Snowflake is one of the top contenders for that. If you are wondering why Snowflake is so popular, look at the user satisfaction — it’s over the roof. Snowflake took most of the drawbacks and oversights that other data storage solutions had and addressed it. Probably the most important feature is having the storage and compute separately scaled.

What is Kafka Connect, and why did we decide to use it?

Kafka Connect is a tool for scalably and reliably streaming data between Apache Kafka and other data systems. It provides streaming integration between Kafka and other data sources or sinks, enabling seamless data movement and transformation. Kafka Connect uses connectors to ingest data into Kafka topics or export data from Kafka topics to external systems.

Kafka Connect is made for use cases like this.

Installing Kafka Connect and choosing the correct mode

If you search for Kafka Connect, you will see that there is no download link. That is because Kafka Connect is not a standalone tool; it is bundled with the main Kafka installation. If you communicated with Kafka from the command line before, you probably used one of the scripts from the directory like kafka-topics.sh . Two more scripts that are used for Kafka Connect are in the same directory as the kafka-topics.sh script. Those two scripts are for two different modes, standalone and distributed:

1. connect-standalone.sh — Only runs one node of Kafka Connect and saves all the configs, offsets, and statuses locally. If the node goes down, all Kafka Connect processes are down.
This is okay for testing but should not be used in production as it is neither reliable nor fault-tolerant.

2. connect-distributed.sh — Runs a cluster of Kafka Connect nodes. Saves all configs, offsets and statuses in specific Kafka topics designated for Kafka Connect. If one node goes down, another node will take over the workload from that node.

The connect-standalone.sh script can be used for testing, but connect-distributed.sh should be used for production.

Kafka Connect components

There are a few Kafka Connect components, but the most important for us are:

1. Connectors (sink and source)
2. Converters (including transforms)
3. Dead Letter Queue (DLQ)

If you are new to Kafka Connect, you may be confused about the difference between connectors and converters.

Connectors

Connectors, as the name says, connect your Kafka Connect cluster to a specific source/destination. In our case, that will be but it can also be HDFS, S3, Elasticsearch, and many more.

There are two types of connectors:

1. Source Connectors — Take the data from a source and send it to Kafka topic(s). In the diagram below, we can see how it would look if we wanted to take files from S3 and publish them to a Kafka topic.

2. Sink Connectors — Take the data from Kafka topic(s) and send it to a destination. In the diagram below, we can see how it would look if we wanted to take messages from a Kafka topic and store them in Snowflake.

Converters

Converters are the most confusing component to set up because it’s easy to misunderstand and misconfigure them. Since Kafka messages are just bytes, converters translate data between the internal data format used by Kafka Connect and the format specific to the source or sink connector. When we store messages in Kafka topics they will always get serialized, even if we do not specify the serialization format. The Kafka producer will use the String serializer by default to serialize the message and save it to a topic as an array of bytes. Usually, you will use serialization formats like Avro or Protobuf for more complex pipelines. Since we are using both connectors and converters, let’s refer to them together as a Kafka Connect system to make it less confusing.

We have two systems: a source system and a sink system. You may see that in some places, those two components are both referred to as connectors, but that is why many people get confused. We will expand both previous diagrams to demonstrate how both source and sink systems look.

Source system

As you can see in the diagram, the source connector (in our case S3 connector) establishes the connection to S3 and starts polling the desired file. After the S3 connector gets a file, it will hand out the contents of that file to the converter. The converter will take that content and serialize it with the specified serializer to the specified format. After serialization, it will publish the serialized message to a Kafka topic.

Sink system

In sink systems, the converter first takes the serialized message from a Kafka topic, deserializes it, and hands it over to the sink connector. After that, the sink connector (in our case the Snowflake connector) saves the formatted message into the Snowflake table. If you are still confused, imagine that the connector connects Kafka Connect to a source/destination and acts as a bridge between those two points while the converter does all the data transformations (serialization/deserialization).

Understanding converters is a crucial part of setting up Kafka Connect. I highly recommend spending some time to understand connectors and converters before attempting to set up Kafka Connect for a complex pipeline. It will save you a lot of time in the long run.

Confluent published a great blog about converters here: Kafka Connect Deep Dive — Converters and Serialization Explained | Confluent.

Dead Letter Queue (DLQ)

A Dead letter queue or DLQ is a component that can be easily overlooked since it is not a required component for setting up Kafka Connect. In my opinion, DLQ is a must for any production grade data pipeline. DLQ is a component that is responsible for handling messages that failed to be consumed. In terms of Kafka, DLQ is a separate topic that is only used to catch messages that had some issues and could not be consumed.

The most common issue when this happens is when using the wrong converter. If you have worked with Avro, there is a good chance you saw the “Unknown magic byte!” error. That error occurs when you try to use an Avro converter to deserialize messages that are not Avro. A wrong converter is one example, but there can be more issues. To avoid losing messages, we need a DLQ that will handle those cases. By default, the Kafka Connect cluster will stop if it cannot deserialize a message, and we will need to manually fix the issue and restart it.

In production environments we cannot allow downtime nor lose messages and that’s where DLQ can help us. We can configure Kafka Connect to not crash when a message cannot be consumed or deserialized but instead send it to the DLQ topic.

These are the properties that we need to define if we want to do that:

"errors.tolerance": "all", 
"errors.deadletterqueue.topic.name": "MY_DLQ_TOPIC",
"errors.deadletterqueue.topic.replication.factor": 3,
"errors.deadletterqueue.context.headers.enable": true

Schema registry

Schema registry is an additional service that helps us improve the data quality of our Kafka data. It’s a centralized repository where we can store our schemas for our Kafka messages. You can think of schemas as blueprints for how the message should look like. Schema definition contains all the fields and their types that a message needs to have. When a producer wants to publish a message to a topic, it must first validate whether the message satisfies the schema definition. Only then can the message be published. The same applies for consumers. This helps us have standards, and establishes better collaboration across different teams, and helps prevent mistakes and inconsistencies.

This is an example how a schema definition looks like:

{
  "doc": "Sample schema",
  "fields": [
    {
      "doc": "ID of the user. Type Int.",
      "name": "id",
      "type": "int"
    },
    {
      "doc": "Users full name. Type String.",
      "name": "name",
      "type": "string"
    }
  ],
  "name": "sampleRecord",
  "namespace": "com.dk.mynamespace",
  "type": "record"
}

Kafka Connect is made by Confluent and has many integrations with the Confluent Platform, and also different licenses. To reduce the complexity of the blog, in part 2 we will use the Confluent Platform schema registry.

In case you cannot use the Confluent Platform, Confluent provides the schema registry under Community License and in 99% of the cases you will be fine to host your own instance of the schema registry. The only exception is if you want to offer it as a service and compete with Confluent. In those cases, feel free to use alternatives like Karapace.

Conclusion

When it comes to streaming messages from Kafka to Snowflake, Kafka Connect was a clear winner for us. It provided us with high reliability, scalability and fault tolerance while also having out of the box integration with schema registry and dead letter queues.

In this blog, we learned about Kafka Connect, how to use it, and which features are necessary for a production-grade big data pipeline.

For a practical guide on how to set up Kafka Connect; stay tuned for part 2 of the blog where we configure and deploy three Kafka Connect clusters (avro, protobuf and string) on Kubernetes and demonstrate the process how we can utilize Schema Registry and DLQ (Dead Letter Queue) to establish a production grade data pipeline.

The architecture we will set up is shown here:

Originally published at https://www.atlantbh.com on May 16, 2024.

Consuming and Storing Kafka Messages in Snowflake Using Kafka Connect (Part 1) was originally published in Atlantbh Engineering on Medium, where people are continuing the conversation by highlighting and responding to this story.

In the previous blog, we first defined the problem of customer support ticket classification that Atlantbh had the opportunity to solve. After presenting the business goal, we briefly described the proposed approach and obtained results. This blog post serves as a follow-up. We aim to delve into individual steps of the proposed approach, highlighting the best practices and providing insights gained from our experience.

Note that although we will remain in the context of customer support ticket classification, most of the steps described below apply to almost any text classification task, regardless of the specific problem domain.

Figure 4. in the previous blog post shows the overview of text classification flow using the ML approach. This figure outlines six steps, from problem definition and business goal to classifier deployment. On the other hand, the figure below shows a more granular flow of text classification by splitting feature engineering and model construction phases into multiple individual steps.

Exploratory Data Analysis

Exploratory Data Analysis (EDA) holds significant practical value for any ML task, providing insight into underlying structure and data characteristics. EDA enables the identification of the class distribution and detection of a potentially imbalanced dataset. However, an imbalanced dataset does not necessarily mean a “red flag”. In practice, imbalanced datasets are quite common. This is also the case with our dataset. For example, customers are more likely to report shipping issues than their account settings issues. There are many different ways to “combat” an imbalanced dataset, such as oversampling or undersampling, and many others, but we will not explain each of these techniques in detail here. Nevertheless, it is important to point out that in the case of an imbalanced dataset, selecting the appropriate metrics for model performance evaluation is of great importance. This will be discussed further as we go through the model construction process.

EDA can also provide valuable information and basic statistics of text corpus (i.e., unstructured text dataset), such as discovering the various data structures within a dataset. It is important to note that EDA is not a one-time process performed during the project’s initial phase, as some may assume by looking at the simplified version of the text classification flowchart. (Figure 1.) On the contrary, there is a feedback loop between EDA and feature engineering steps. (Figure 2.)

Insights gained from the initial EDA directly influence the selection of steps we will perform in the feature engineering phase. Let’s say that we discover through EDA that there are some irrelevant records — it is necessary to remove these records from the dataset in the data cleaning phase. Conversely, the feature engineering steps also affect EDA. As the data goes through various changes in the feature engineering phase, it prompts a re-examination of the EDA to assess the impact of the feature engineering on the dataset and yield new insights. For example, stop word removal or lemmatization changes the previously established characteristics of the text corpus, and it is necessary to re-establish them. This feedback loop between EDA and feature engineering allows the development of a robust text classification model that increases the effectiveness of capturing underlying patterns in text data.

Feature Engineering Phase

Several fundamental steps are often performed during the feature engineering phase of a text classification task to ensure the quality and relevance of features extracted from raw text data. However, steps such as language translation or certain data cleaning and text preprocessing techniques may vary depending on the dataset characteristics and task-specific requirements. For example, language translation is crucial when dealing with multilingual datasets but is irrelevant for monolingual ones. Similarly, if a dataset contains different data formats, additional steps of data cleaning and text preprocessing may be required to resolve this variability. By recognizing a balance between commonly applicable procedures and task-specific requirements, we can tailor our feature engineering strategies to effectively address the unique challenges given by each NLP task and dataset.

Data Cleaning

After performing EDA on the initially obtained data, we quickly realized that our dataset contains records in various formats. In addition to plain text, which accounts for most of the dataset, we concluded that around 10% of records represent inquiries in HTML format. In many real-world situations, this often happens. For example, when collecting data from websites or emails, text usually contains HTML tags, which are irrelevant for analysis and may even interfere with further preprocessing steps. Removing these HTML tags and converting such text into plain, readable format was one of the first steps in data cleaning. Beautiful SoupPython library has proven to be a powerful tool to efficiently identify HTML-like messages, while the html2text library demonstrated its effectiveness in converting these messages to plain text.

The second part of data cleaning involved identifying and removing duplicates and irrelevant or noisy records from the dataset to ensure the data quality. In our scenario, the dataset included different types of non-standard records, such as test messages or other forms of inquiries that do not contribute to the classification objective, so it was necessary to develop specific conditions for filtering such records. Manual inspection also enabled us to verify and flag potentially irrelevant records. Once identified, such records were excluded from further analysis during the data cleaning process to prevent them from influencing the model training process and compromising classification performance.

Data cleaning usually reduces the dimensionality of the initial dataset by removing irrelevant features and records. In general, by filtering unnecessary columns (i.e., features) and discarding rows (i.e., records), the dataset becomes more streamlined. However, one should be very careful not to overdo the cleaning process, as excessive data removal can result in the loss of potentially valuable information, which can also negatively affect the performance of the classification model. Therefore, finding the right balance between the need for data cleaning and the preservation of relevant data is essential.

Language Detection & Translation

In the customer support domain, where inquiries may be in different languages, translation is necessary when creating a classification model. Instead of creating different models for each language, using language detection and translation tools is a more cost-effective and pragmatic approach. This strategy consolidates inquiries in multiple languages into a single dataset, facilitating comprehensive analysis. Since we worked with the data of an international company, it is not surprising that our dataset is multilingual. More precisely, our dataset contains customer inquiries in over 20 languages, where about 13% of records are in a language other than English.

When using APIs for language detection and translation, it is critical to consider data confidentiality and security. Ensuring that sensitive information remains protected during the translation is paramount and requires a careful selection of APIs and robust security protocols to protect data integrity and privacy.

Text Preprocessing

As already mentioned, text preprocessing steps are highly dependent on the specifics of the task and the dataset we are working with. In-depth EDA typically yields an intuitive determination of the necessary text preprocessing steps. We used the power of the Natural Language Toolkit (NLTK) Python library to perform around 15 individual steps as part of the text preprocessing phase. This library is popular and widely used for various NLP tasks. Performed steps range from converting all text to lowercase to removing emojis, URLs, special characters, irrelevant phrases, stop words, and many more. (Figure 3.)

Here, we will focus on two crucial steps in almost every text classification task: lemmatization and tokenization.

Lemmatization

Lemmatization is a text preprocessing technique that reduces words to their base or root forms (i.e., lemmas). Applying this preprocessing step leads to text normalization and vocabulary dimensionality reduction. There is another similar text preprocessing technique called stemming. Stemming removes suffixes to find the word root form, which does not necessarily result in valid words. On the other hand, lemmatization considers word context and part of speech, resulting in more precise transformations that represent valid words. (Figure 4.) However, although lemmatization is a more sophisticated approach than stemming, the downside is that it is more computationally intensive and much slower than stemming, which should be kept in mind when working with high-dimensional text datasets.

Tokenization

The term “tokenization” without specifying any additional context refers to the word tokenization. It is the most commonly used form of tokenization and involves splitting a text into individual words or tokens. However, we should note that there are also other types of tokenization. For example, NLTK also implements the sentence tokenization method. By breaking down text into smaller chunks, tokenization enables more granular analysis and further text data processing. Tokenization is a fundamental technique in every text classification task, as it serves as the preliminary step in converting raw text data into a format suitable for analysis by ML algorithm.

Train-Test Split

After tokenization, it is time to split the dataset into training and test subsets. This step is always better to perform before vectorization, especially when training our vectorization model rather than using a pre-trained one. We leave the test dataset aside and use it only at the very end to evaluate the performance of the classification model. This way, we ensure that the training and test datasets remain independent, thus preventing any information leakage from the test to the training dataset.

In general, the train-test split ratio always hovers around 80/20. This ratio has its roots in the well-known Pareto principle (also known as the 80/20 rule), which states that approximately 80% of consequences come from 20% of causes (the “vital few”). We decided to split our dataset using a 75/25 ratio. In the case of an imbalanced dataset like ours, it is necessary to provide an equal distribution of classes in the training and test datasets. This way, we ensure that the minority class is not included exclusively in the training or test dataset. Fortunately, this can be easily achieved using the “shuffle” and “stratify” arguments of sklearn’s train_test_split method.

Vectorization

Since ML algorithms cannot directly process text, conversion of text to numerical representation is an essential part of every text classification task. This process is called vectorization and is the final step in the feature engineering phase. We decided to use Word2Vec and train it on our data instead of directly using a pre-trained model. For training the Word2Vec model, we used the power of the Gensim library. This Python library has a module that efficiently implements the Word2Vec family of algorithms.

Once trained, Word2Vec is used on both training and test datasets to generate word embeddings for each word in the dataset vocabulary. This way, words are represented as high-dimensional vectors of numbers optimized for our domain-specific semantics. At the same time, Word2Vec ensures that relationships between words are captured so that similar meanings or contexts have similar vector representations.

Given that the previously tokenized dataset contains customer inquiries of variable length, word embeddings are first generated for each word and then aggregated to obtain a single vector representation for the entire record. This aggregation combines the information from all words in the record into a fixed-length vector determined by the dimensionality specified during the training of Word2Vec model. One of the most frequently used methods of aggregating is averaging word embeddings. (Figure 5.) This fixed-size vector of numbers captures the semantic information within the record and represents a suitable input to ML models.

Model Construction Phase

The model construction phase, which follows feature engineering, consists of six steps (Figure 1.). Although these steps are closely related, they are presented separately on the flowchart to underline their importance. In the following sections, we will clarify these steps by emphasizing two decisions: classification algorithm selection and evaluation metrics.

Classification Algorithm Selection

Choosing the appropriate classification algorithm is a vital decision that is heavily influenced by the data scientist’s experience and domain knowledge. Experienced data scientists use their understanding of the dataset characteristics, the nature of the problem, and the known strengths and limitations of various algorithms to decide which algorithm to use. However, experimenting with different classification algorithms is often convenient as it provides valuable insights into which algorithms work best for a given task.

After evaluating the strengths and weaknesses of various algorithms, we chose Support Vector Machine (SVM) as the classification algorithm for its proven effectiveness in different NLP tasks. SVM’s several advantages make it a compelling choice for text classification tasks. Firstly, SVM is well-known for its versatility and robustness. It can handle high-dimensional data and capture complex relationships. SVM has the ability to find the optimal hyperplane for separating classes in a feature space, even in cases where the data is not linearly separable. It achieves this by using different kernel functions (e.g., linear, polynomial, radial basis functions). Moreover, SVM’s regularization parameter allows for fine-tuning the trade-off between model complexity and generalization. This flexibility ensures avoiding overfitting while maintaining good performance on unseen data.

Firstly, the baseline SVM model is trained and evaluated to establish a performance benchmark. The performance of this baseline model is then used to estimate the level of improvement in subsequent iterations. Hyperparameter tuning enabled us to determine the optimal SVM model parameters (e.g., kernel type, regularization parameter, and gamma value). More precisely, hyperparameter tuning entails multiple iterations of model training and evaluation using different combinations of hyperparameters. Once the model iterations are complete, the best-performing SVM model is selected. In the case of experimenting with several classification algorithms, the previously described steps are performed for each selected algorithm. Finally, by comparing each of them, the final model is chosen.

It is important to note that if we are not satisfied with the obtained results, we should consider going back to EDA, reviewing the steps performed in the feature engineering phase, modifying some of the preprocessing steps, or perhaps considering using a different algorithm to generate word embeddings.

Evaluation Metrics

For assessing the performance of a classifier, choosing the right evaluation metrics, especially in the context of imbalanced datasets, is vital. Stakeholder input becomes crucial as their priorities and goals often determine evaluation metrics. In our scenario, the stakeholders’ input primarily focused on accuracy as the key metric. However, we also relied on class-wise metrics such as precision, recall, and F1-score to ensure a comprehensive assessment. In addition, by using micro, macro, and weighted averages, we gained overall insights into model effectiveness.

Classifier Deployment

After the final model selection, the classifier is deployed in the production environment. Applying the created classifier in production on new customer support tickets (i.e., unseen text) consists of several steps. Firstly, it is necessary to detect the language of the new text and translate it if it is not in English. After that, the exact text preprocessing steps included in the process of classifier construction follow. Finally comes vectorization, which converts the text into a form suitable as input to the created classifier. By applying the classifier, the output is the category of a customer support ticket.

In the classifier deployment phase, cooperation between different teams is crucial. Software engineers developing the system are responsible for the seamless classifier integration, while DevOps engineers manage the deployment pipeline and infrastructure. Data scientists who build the model work alongside them to ensure a smooth transition and continuous monitoring of the deployed classifier.

Proper implementation of MLOps practices ensures continuous integration, deployment, and monitoring of classifier performance. In addition, client expectations regarding the frequency of retraining should be clearly defined and agreed upon to maintain the model’s accuracy and relevance over time. This collaborative effort and adherence to best practices in implementation are essential to achieving sustainable and efficient ML-based solutions in real-world applications.

Originally published at https://www.atlantbh.com on April 9, 2024.

Blog by Anja Plakalović, Data Analyst at Atlantbh.

Comprehensive Guide: Creating an ML-Based Text Classification Model was originally published in Atlantbh Engineering on Medium, where people are continuing the conversation by highlighting and responding to this story.

Natural Language Processing (NLP), one of the branches of AI, encompasses a powerful set of techniques with a wide range of applications, especially in the field of customer support. By enabling machines to understand, interpret, and generate human language, NLP breaks down the barriers between technology and human interaction, paving the way for a more intuitive user experience and personalized customer service.

Whether you are an experienced organization or a budding entrepreneur, it is important that you understand how NLP can be used to improve your business. Join us as we explore NLP by practically applying it to one common task. In addition, discover the success story of the collaboration of two Atlantbh teams, the development team and a small group of people from the data team, which led to the construction of a customized text classification model and its smooth deployment in production.

Using NLP in Customer Service

NLP tasks play an important role in automating and optimizing various aspects of customer service, improving the overall user experience and operational efficiency. More and more businesses are developing conversational agents (chatbots) that can interact with customers in natural language, respond to frequently asked questions, or solve basic inquiries. Text summarization techniques can be applied to automatically generate concise summaries in cases of rather long customer inquiries. Speech recognition techniques enable transcribing phone calls or voicemails into text. Real-time translation services provide multilingual customer support and enable smooth communication with customers who speak different languages.

There are numerous examples of using different NLP tasks in customer support, but text classification is certainly one of the most common. Problems such as spam filtering, sentiment analysis, ticket triaging, or ticket classification can be considered as variations of the same task: text classification.

Figure 1. Examples of Text Classification in Customer Service (Anja Plakalovic)

Problem Definition: Customer Support Ticket Classification

For any business, customer complaints are important, as they can often indicate shortcomings in their products or services. If these complaints are not resolved quickly, it can lead to customer dissatisfaction, while a recurring trend of dissatisfaction can lead to reduced revenue.

Therefore, solving the problem of automatic classification of customer support tickets has a significant value, both for businesses and for customers. From a business perspective, it leads to a streamlined customer support workflow and improved operational efficiency. On the other hand, from a customer perspective, it ensures that their inquiries are immediately routed to the appropriate team, resulting in faster responses and a smoother user experience.

Atlantbh had the opportunity to work on solving the problem of classification of customer support tickets for an international company. We can formally define this problem as follows:

Given a set of customer support tickets (i.e., formal records of customer inquiries, issues, or complaints) and a predefined set of categories (e.g., shipping issues, refund issues, account settings issues), the task is to construct a model that can accurately classify each ticket into the appropriate category, depending on the ticket content.

Business Goal

The existing process for handling customer support tickets relied entirely on manual categorization, where customers were tasked with selecting a specific category before submitting a problem. However, this approach has proven to be error-prone, with frequent occurrences of misclassification. Consequently, customer inquiries were often routed to the wrong support teams, leading to delays in resolution and customer frustration. In addition, support teams spent significant time recategorizing misclassified tickets and routing them to the appropriate teams. This manual reassignment process increased response times and resulted in redundant efforts and inefficiencies across various support teams.

Figure 2. Customer Support Ticket Handling Process: Before vs. After (Anja Plakalovic)

The proposed support ticket handling process intends to automate ticket categorization using a custom text classification model. This solution eliminates the need for customers to manually select ticket categories by using NLP and ML techniques to automatically classify tickets based on their content. This also simplifies the user interface for reporting problems, allowing users to report problems more quickly without having to decide which category the problem belongs to. (Figure 3.)

By implementing this proposed approach, the goal is to achieve a considerable improvement in the accuracy and efficiency of support ticket classification, surpassing the estimated accuracy of the existing solution, which is approximately 80%. This way, we aim to reduce the time teams spend recategorizing misclassified tickets and improve overall customer satisfaction by providing faster solutions to their inquiries.

Figure 3. Report a Problem Form: Before vs. After (Anja Plakalovic)

Approach

The client first approached the development team with a request to create a solution for automatically classifying support tickets. After defining the problem, the development team extracted data from the ticketing system. This way, a labeled dataset was collected, containing the text content of existing tickets (i.e., customer inquiries) and each ticket’s corresponding category or label.

The development team tried to solve the problem by applying a rule-based approach. This method is often the initial and simplest strategy for solving text classification problems. A rule-based approach uses predefined rules to classify support tickets based on specific criteria or linguistic patterns. However, despite the team’s efforts, it quickly became apparent that the rule-based approach alone could not adequately address the complexity and variability of ticket content. As a result, the team recognized the need to explore more advanced methodologies, which led to collaboration with the data team and using ML approach to solve this problem.

The figure below shows an overview of creating a text classifier using a supervised ML approach, delineating the teams involved and their respective responsibilities. Considering that we already clarified the first two steps, we will henceforth describe the remaining steps included in the used approach.

Figure 4. Overview of Text Classification Flow Using ML Approach (Anja Plakalovic)

The data team’s first step after being introduced to the problem by the development team was to familiarize themselves with the data and develop a sense of what could be derived from it. This step is formally called Exploratory Data Analysis (EDA). It enabled us to raise awareness of potential challenges and constraints early in the project lifecycle, guiding the appropriate further approaches and risk mitigation. In the context of our customer support ticket classification problem, EDA provided valuable insights regarding customer inquiries. These, among others, include the distribution of ticket categories, the frequency of specific keywords or phrases within each category, and the customer inquiry length distribution.

A thorough EDA created a good foundation for defining steps in the feature engineering phase, which ensured converting the raw text data into a format suitable for the ML classification algorithm. Some of the performed feature engineering steps include data cleaning, standard text preprocessing techniques such as stop word removal, lemmatization or tokenization, and text vectorization at the end. Making the strategic decision to use Word2Vec as a text vectorization algorithm ensured we captured semantic relationships and context within the text. Rather than relying on pre-trained models, training Word2Vec on our dataset provided word embeddings tailored to our use case. Before performing text vectorization, we split the dataset into training and test datasets and trained Word2Vec only on training data. Afterward, we saved the trained Word2Vec model to preserve the learned word embeddings and use it to vectorize training and test data.

Upon completing the transformation of our raw customer inquiries into a suitable format as input to the ML algorithm, the next step entailed selecting a suitable classification algorithm. Extensive research encompassing a spectrum of classification algorithms, complemented by our profound domain knowledge acquired through the EDA phase and an exhaustive feature engineering endeavor, played a crucial role in this decision-making process. The culmination of these efforts led us to discern the Support Vector Machine (SVM) as the optimal choice for our text classification task. This decision was reinforced by SVM’s well-documented prowess in handling high-dimensional and sparse data since these characteristics are common in text classification tasks. Its ability to delineate complex decision boundaries, robustness in handling non-linear relationships, and good generalization performance further solidified its importance in our use case. By implementing hyperparameter tuning, we ensured using the optimal SVM parameters that improve classification performance. After training the model with these parameters on the training dataset, we saved the SVM model. Then, we used it on the test dataset to evaluate the performance of the created model.

It is interesting to consider the distribution of time that the Atlantbh data team invested in implementing the previously mentioned phases. It is important to note that the following results represent rough estimates and depend significantly on the nature of the problem and the specific use case. In our scenario, we invested approximately 10% of the total time in EDA. This effort mainly included a thorough familiarization with the dataset at the beginning of the project. On the other hand, the feature engineering phase took the majority share with 60%, showing significant time investment in the data preprocessing before the model training. Model construction, which included ML models’ development and refinement, accounted for the remaining 30%.

Figure 5. Time Allocation Overview: EDA, Feature Engineering, and Model Construction (Anja Plakalovic)

Results

By implementing the proposed approach, our primary goal was to significantly improve the accuracy and operational efficiency of the customer support ticket classification system. We exceeded the estimated accuracy of the existing solution, which was around 80%. More precisely, our approach resulted in an outstanding performance, achieving 93% accuracy when evaluated on the test dataset. After presenting the results to our clients, diligent work and seamless cooperation of different Atlantbh teams enabled the smooth deployment of the proposed model to production. A few months after the model deployment, we evaluated its accuracy in production. It turned out that in production, the proposed model for customer support ticket classification has an accuracy of an impressive 98%, which significantly exceeds even its initially estimated accuracy.

This significant improvement not only confirms the effectiveness of our approach but also highlights the commitment and expertise of everyone involved in the process. By surpassing the previous performance, we effectively reduced the time spent recategorizing misclassified tickets, streamlined operational flows, and increased team productivity. Moreover, this improved accuracy further leads to faster and more accurate responses to customer inquiries, which increases customer satisfaction. This success underlines the significant benefits and improvements businesses can achieve using NLP and ML techniques, and Atlantbh’s commitment to delivering exceptional results and creating tailored solutions.

If you found this blog engaging, we encourage you to read part two: “Comprehensive Guide: Creating an ML-Based Text Classification Model”. As the name suggests, this blog provides an in-depth description of individual steps of the proposed approach.

Originally published at https://www.atlantbh.com on April 9, 2024.

Blog by Anja Plakalović, Data Analyst at Atlantbh.

Harnessing the Potential of NLP: Effortless Experience & Efficient Customer Service was originally published in Atlantbh Engineering on Medium, where people are continuing the conversation by highlighting and responding to this story.

Test Automation Integrated into AWS Step Function Workflow

AWS Step Function is an AWS resource that offers serverless function orchestration making it easy to sequence multiple AWS resources into one business workflow. This tool allows us to create and run many checkpointed and event-driven workflows that maintain the application state, ideal for projects such as data pipelines and many others. In this blog, we will explain the concept of development and application of the Step Function, with test automation integrated into its workflow.

Pricing of AWS Step Function

AWS Step Function pricing works based on ‘you pay what you use’. We have two types of workflows: Standard Workflows and Express Workflows. By default, Step Function uses Standard Workflow, and its price is $0.000025 per state transition. The pricing for Express Workflow is $0.000001 per request. In this blog, we will use the default option.

Step Function states

AWS Step Function provides a set of states that you can use to build workflows. Each state represents a specific action or behavior within the workflow. Here are some of the commonly used Step Function states:

1. Task State: Represents a single unit of work in a Step Function workflow. It can be used to invoke AWS Lambda functions, run ECS tasks, or perform other activities.
2. Choice State: Allows you to define conditional logic within your workflow. It evaluates a condition and transitions to different states based on the result.
3. Wait State: Delays the execution of the workflow for a specific period.
4. Pass State: Passes its input to its output without performing work. It is useful when constructing and debugging state machines.
5. Parallel State: Allows us to run multiple branches of execution in parallel. Within the parallel state every branch can be separated.
6. Succeed State: Ends execution as successfully processed.
7. Fail State: Ends execution as failure. It also provides more ways to find out the root cause of errors. Additionally, it could provide us with the ‘Catch’ block if necessary.
8. Map State: You can think of it like a for-each loop for Step Function. It allows us to iterate over an array/list of items and perform the same workflow steps for every item in the array.

Let us learn by the example:

Demo project description:

In this example, we will create a simple demo software with the Step Function, Lambda, DynamoDB, GitHub Actions, and Serverless framework for easier Step Function creation. In the DynamoDB, we will create one table called ‘messages’. This table is supposed to receive data from other imaginary services. Message record should contain the following:

• ID,
• First name (fName),
• Last name (lName),
• Region (US/EU),
• Points,
• Column representing flag for processed messages (is_processed)

Our Step function should contain a workflow that takes all of the unprocessed messages, separates them by the region of the US and EU users, and puts those messages into the other appropriate tables: US_users and EU_users.

Step function implementation:

We will use the Serverless framework to easily implement the Step Function, its Lambdas and IAM roles. The Serverless framework offers infrastructure as a code implementation and deployment for the AWS Step function. It uses Lambdas and additional necessary services that we need in the state machine. We will not go in-depth with the Serverless framework, but if you would like to know more, you can find it on their official website.

Follow these steps to start with the Serverless framework:

1. Create empty folder

2. Instantiate NPM with the ‘npm init’ command

3. Install serverless framework via NPM ‘npm install -g serverless’

4. Add the following dependencies that we will further use in package.json:

{
  "name": "sfn-blog",
  "dependencies": {
    "@aws-sdk/client-dynamodb": "^3.405.0",
    "@aws-sdk/client-s3": "^3.405.0",
    "axios": "^1.4.0",
    "jest": "^29.4.1",
    "jest-junit": "^15.0.0"
  },
  "version": "1.0.0",
  "description": "",
  "main": "index.js",
  "directories": {
    "test": "test"
  },
  "scripts": {
    "test": "jest"
  },
  "author": "",
  "license": "ISC",
  "devDependencies": {
    "@types/jest": "^29.2.5",
    "serverless-step-functions": "^3.13.1"
  }
}

5. Run ‘npm install’

6. Create serverless.yml which will be used for infrastructure as a code

7. Create for lambda functions written in Node JS

Note: Do not forget to add your AWS credentials (access key and secret key). If you are on a MAC machine, you can do this by running the following command:

sls config credentials --provider aws --key <YOUR_ACCESS_KEY> --secret <YOUR_SECRET_KEY>

8. Create DynamoDB table ‘messages” with ID type String as a Partition key:

9. Choose Customize Settings, On-demand capacity mode, and click ‘ Create table ‘ at the end of the page.

10. In the same way, create two more tables: US_users_events and EU_users_events.

11. In the serverless yml file, we will define our needed Step Function workflow:

service: sfn-blog
frameworkVersion: "3"
# Define  cloud provider settings and IAM roles needed for our SF to work
provider:
  name: aws
  runtime: nodejs18.x
  region: eu-central-1
  environment: ${file(env.json)}
  iamRoleStatements:
    - Effect: Allow
      Action: dynamodb:*
      Resource: arn:aws:dynamodb:eu-central-1:178190218027:table/messages
    - Effect: Allow
      Action: dynamodb:*
      Resource: arn:aws:dynamodb:eu-central-1:178190218027:table/US_users_events
    - Effect: Allow
      Action: dynamodb:*
      Resource: arn:aws:dynamodb:eu-central-1:178190218027:table/EU_users_events
    - Effect: Allow
      Action: s3:*
      Resource: arn:aws:s3:::sfn-blog
plugins:
  - serverless-step-functions
# Define a path to the AWS Lambda Functions
functions:
  FetchFromDynamoDBState:
    handler: handler.FetchFromDynamoDBState
  ProcessUSUsersEvents:
    handler: handler.ProcessUSUsersEvents
  ProcessEUUsersEvents:
    handler: handler.ProcessEUUsersEvents
  InsertMessageDynamoDBTest:
    handler: handler.InsertMessageDynamoDBTest
  VerifyStepFunctionOutcomeTest:
    handler: handler.VerifyStepFunctionOutcomeTest
# Define Step Function and its state machine
stepFunctions:
  stateMachines:
    proceedRewards:
      name: proceedRewards
      definition:
        StartAt: InsertMessageDynamoDBTest
        States:
          InsertMessageDynamoDBTest:
            Type: Task
            Resource:
              Fn::GetAtt: [InsertMessageDynamoDBTest, Arn]
            ResultPath: "$.response"
            Next: WaitState
          WaitState:
            Type: Wait
            Seconds: 200
            Next: FetchFromDynamoDBState
          FetchFromDynamoDBState:
            Type: Task
            Resource:
              Fn::GetAtt: [FetchFromDynamoDBState, Arn]
            ResultPath: "$.items"
            Next: ProcessDataState
          ProcessDataState:
            Type: Map
            ItemsPath: "$.items.items"
            ResultPath: "$.mappedData"
            MaxConcurrency: 2
            Iterator: 
              StartAt: ProceedChoiceState
              States:
                ProceedChoiceState:
                  Type: Choice
                  Choices:
                    - Variable: $.region.S
                      StringEquals: "US"
                      Next: ProcessUSUsersEventsState
                    - Variable: $.region.S
                      StringEquals: "EU"
                      Next: ProcessEUUsersEventsState
                  Default: DefaultState
                ProcessUSUsersEventsState:
                  Type: Task
                  Resource:
                    Fn::GetAtt: [ProcessUSUsersEvents, Arn]
                  End: true
                ProcessEUUsersEventsState:
                  Type: Task
                  Resource:
                    Fn::GetAtt: [ProcessEUUsersEvents, Arn]
                  End: true
                DefaultState:
                  Type: Fail
                  Cause: 'Invalid region value.'
                  Error: 'InvalidRegionError'
            Next: VerifyStepFunctionOutcomeTest
          VerifyStepFunctionOutcomeTest:
            Type: Task
            Resource:
              Fn::GetAtt: [VerifyStepFunctionOutcomeTest, Arn]
            End: true

12. Include the following Lambda functions in the handler.js file (if you are wondering why we are using these keys ‘S’, ’N’ and ‘BOOL’ in params for DynamoDB client it is because the library itself needs that specification of the data types, ‘S’ for string, ’N’ for number, ‘BOOL’ for boolean and etc.):

const { DynamoDBClient, PutItemCommand, ScanCommand, UpdateItemCommand } = require('@aws-sdk/client-dynamodb');
const dynamoDbClient = new DynamoDBClient({ region: 'eu-central-1' });
module.exports.FetchFromDynamoDBState = async () => {
try {
const params = {
            TableName: 'messages',
            FilterExpression: 'is_processed = :processed',
            ExpressionAttributeValues: {
':processed': { BOOL: false },
            }
        };
const scanCommand = new ScanCommand(params);
let result = await dynamoDbClient.send(scanCommand);
const items = result.Items;
return {
            items
        };
    } catch (error) {
throw error;
    }
}
module.exports.ProcessUSUsersEvents = async (item) => {
return await processUsersEvents(item, 'US_users_events');
}
module.exports.ProcessEUUsersEvents = async (item) => {
return await processUsersEvents(item, 'EU_users_events');
}
const processUsersEvents = async (item, tableName) => {
let dataProcessed = false;
try {
await processItem(item, tableName);
await updateIsProcessedColumn(item.ID.S);
        dataProcessed = true;
    } catch (error) {
return dataProcessed;
    }
return dataProcessed;
}
const processItem = async (item, tableName) => {
const eventItem = {
        ID: { S: item.ID.S },
        fName: { S: item.fName.S },
        lName: { S: item.lName.S },
        points: { N: item.points.N },
    }
const params = {
        TableName: tableName,
        Item: eventItem
    };
const command = new PutItemCommand(params);
await dynamoDbClient.send(command);
}
const updateIsProcessedColumn = async (itemId) => {
const paramsForUpdate = {
        TableName: 'messages',
        Key: {
            ID: { S: itemId },
            UpdateExpression: 'SET is_processed = :value',
            ExpressionAttributeValues: { ':value': { BOOL: true } },
            ReturnValues: 'UPDATED_NEW'
        }
    };
const updateItemCommand = new UpdateItemCommand(paramsForUpdate);
await dynamoDbClient.send(updateItemCommand);
}13. Run command: ‘sls deploy’ to the Step Function:

Test Automation

In real-world scenarios, this Step Function can be scheduled to run a workflow at regular intervals, such as every hour, so we have to split the test automation into two parts: data insertion and verification.

Firstly, we will create a test automation script to insert a message with test data to the DynamoDB table ‘messages’ and save this message as a test job artifact to an S3 bucket.

AWS S3 is a cloud-based object storage service provided by Amazon Web Services (AWS) (you can learn more here ) that we need in this case to store our test artifacts that will be later used from the second script that will do the verification part. You can think of the S3 bucket as a virtual container where our test data will be kept, but of course, you can use other services for this purpose. The good alternatives could be saving the data as the GitHub artifacts, or if you choose to work this with Jenkins, you can have the same concept and also save the data on your own servers. It is also important to explain that we can not save the artifacts locally on AWS Lambdas because they are serverless computing services that are packaged into isolated environments (containers) where it runs on a multi-tenant cluster of machines managed by AWS.

We will use Node JS with JEST for this example.

const { DynamoDBClient, PutItemCommand } = require('@aws-sdk/client-dynamodb');
const dynamoDbClient = new DynamoDBClient({ region: 'eu-central-1' });
const S3Client = require('./S3Client');
describe('Adding message to the DynamoDB messages table', () => {
let item;
    beforeAll(async () => {
        item = {
            ID: { S: new Date().toISOString() },
            fName: { S: 'Test_First_Name' },
            lName: { S: 'Test_Last_Name' },
            points: { N: '500' },
            is_processed: { BOOL: false },
            region: { S: 'US' }
        };
    });
    it('Adding message', async () => {
const params = {
            TableName: 'messages',
            Item: item
        };
try {
const command = new PutItemCommand(params);
await dynamoDbClient.send(command);
console.log('Data inserted successfully.');
        } catch (error) {
console.log('Error inserting data:', error);
        }
    }, 10000);
    it('Save message to S3 as a job artifact ', async () => {
const s3Client = new S3Client();
const bucket = 'sfn-blog';
const key = 'test-message.json';
await s3Client.removeObject(bucket, key);
await s3Client.uploadFileToS3(
            bucket,
            key,
JSON.stringify(item),
'application/json'
        );
    }, 20000);
});

Secondly, we will create a script that should be run after the Step Function finishes to verify that the message has been processed. In this script, we will pull the artifact JSON file from S3 bucket, and verify that this record can be found in the ‘US_users_events’ table in DynamoDB.

const { DynamoDBClient, ScanCommand } = require('@aws-sdk/client-dynamodb');
const dynamoDbClient = new DynamoDBClient({ region: 'eu-central-1' });
const S3Client = require('./S3Client');
describe('Verify that message has been processed', () => {
let item;
const s3Client = new S3Client();
    beforeAll(async () => {
        item = await s3Client.getObject('sfn-blog', 'test-message.json');
    }, 200000);
    it('Verify DynamoDB US_users_events', async () => {
const params = {
            TableName: 'US_users_events'
        };
const scanCommand = new ScanCommand(params);
const response = await dynamoDbClient.send(scanCommand);
const records = response.Items;
const testRecord = records.filter(it => it.ID.S == item.ID.S);
        expect(testRecord.length).toBe(1);
        expect(testRecord[0].fName.S).toBe(item.fName.S);
        expect(testRecord[0].lName.S).toBe(item.lName.S);
        expect(testRecord[0].points.N).toBe(item.points.N);
    }, 20000);
});

We also need a helper file for S3 related methods. Create S3Client.js file:

const { S3, DeleteObjectCommand, GetObjectCommand, PutObjectCommand } = require('@aws-sdk/client-s3');
class S3Client {
    s3Client;
constructor() {
this.s3Client = new S3({ region: 'eu-central-1' });
    }
async getObject(bucketName, key) {
try {
const params = {
                Bucket: bucketName,
                Key: key
            };
const getObjectCommand = new GetObjectCommand(params);
const response = await this.s3Client.send(getObjectCommand);
const bodyStream = response.Body;
let data = "";
for await (const chunk of bodyStream) {
                data += chunk;
            }
const object = JSON.parse(data);
return object;
        } catch (error) {
console.log(error);
return null;
        }
    }
async removeObject(bucketName, key) {
try {
const params = {
                Bucket: bucketName,
                Key: key
            };
const removeObjectCommand = new DeleteObjectCommand(params);
await this.s3Client.send(removeObjectCommand);
        } catch (error) {
console.log(error);
return null;
        }
    }
async uploadFileToS3(bucketName, key, body, contentType) {
try {
const params = {
                Bucket: bucketName,
                Key: key,
                Body: body,
                ContentType: contentType
            }
const putObjectCommand = new PutObjectCommand(params);
await this.s3Client.send(putObjectCommand);
        } catch (error) {
console.log(error);
return null;
        }
    }
}
module.exports = S3Client;

We will now create workflows (think of it as a Jenkins job) for these two scripts with GitHub Actions, but you can use any tool you want, for instance, Jenkins, TeamCity, etc. If you want to know more about GitHub Actions, you can find it on their official website.

To define workflows in GitHub Actions, follow these steps:

1. Create a repository on GitHub,
2. Upload test files, including package.json,
3. Click on the ‘Actions’ button
4. Click on the ‘New workflow’ button,
5. Choose ‘Node.js’ workflow,
6. Define your workflow for the first test script as a yml file inside folder ‘.github/workflows’:

name: insert-message-DynamoDB
on:
  workflow_dispatch:
env:
AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
AWS_REGION: eu-central-1
jobs:
  insert_message_in_DynamoDB:
    name: Insert message in DynamoDB
    runs-on: ubuntu-latest
    steps:
    - name: Checkout repository
      uses: actions/checkout@v2
    - name: Set up Node.js
      uses: actions/setup-node@v2
      with:
        node-version: '18'
    - name: Configure AWS credentials
      run: aws configure set aws_access_key_id ${{ env.AWS_ACCESS_KEY_ID }} && aws configure set aws_secret_access_key ${{ env.AWS_SECRET_ACCESS_KEY }} && aws configure set region ${{ env.AWS_REGION }}
    - name: Install dependencies
      run: npm install
    - name: Run InsertMessageDynamoDB tests
      run: npm run test insertMessageDynamoDB.test.js

7. Do the same for the second script:

name: verify-step-function-outcome
on:
  workflow_dispatch:
env:
AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
AWS_REGION: eu-central-1
jobs:
  verify_outcome:
    name: Verify Step Function Outcome
    runs-on: ubuntu-latest
    steps:
    - name: Checkout repository
      uses: actions/checkout@v2
    - name: Set up Node.js
      uses: actions/setup-node@v2
      with:
        node-version: '18'
    - name: Configure AWS credentials
      run: aws configure set aws_access_key_id ${{ env.AWS_ACCESS_KEY_ID }} && aws configure set aws_secret_access_key ${{ env.AWS_SECRET_ACCESS_KEY }} && aws configure set region ${{ env.AWS_REGION }}
    - name: Install dependencies
      run: npm install
    - name: Run VerifyStepFunctionOutcome tests
      run: npm run test verifyStepFunctionOutcome.test.js

8. Add InsertMessageDynamoDBTest as a first step in the Step Function States array, which should invoke the Lambda function that will trigger the Github Action workflow for running the first test script that will insert a message in DynamoDB ‘messages’ table,

9. Add a wait state after state for triggering the first test script so we can be sure that script execution on the GitHub side has been finished

10. Add a state at the end of the Step Function that should invoke the Lambda function that will trigger the Github Action workflow for running the second test case that will verify that the message has been processed successfully

11. Add env.json file with the GitHub Actons credentials and trigger IDs:

{
    "INSERT_ACTION_TRIGGER_ID": <GITHUB_WORKFLOW_ID_FOR_DYNAMODB_TEST_DATA_INSERTION>,
    "VERIFY_ACTION_TRIGGER_ID":<GITHUB_WORKFLOW_ID_FOR_DYNAMODB_TEST_DATA_VERIFICATION>,
    "GITHUB_OWNER": <YOUR_GITHUB_USERNAME>,
    "GITHUB_REPO": <YOUR_GITHUB_REPO>,
    "GITHUB_TOKEN": <YOUR_GITHUB_TOKEN>
}

12. Specify the environment file inside serverless.yml file under the provider configuration

The final version of the serverless.yml file for the Step Function definition should look like the following:

service: sfn-blog
frameworkVersion: "3"
# Define  cloud provider settings and IAM roles needed for our SF to work
provider:
  name: aws
  runtime: nodejs18.x
  region: eu-central-1
  environment: ${file(env.json)}
  iamRoleStatements:
    - Effect: Allow
      Action: dynamodb:*
      Resource: arn:aws:dynamodb:eu-central-1:178190218027:table/messages
    - Effect: Allow
      Action: dynamodb:*
      Resource: arn:aws:dynamodb:eu-central-1:178190218027:table/US_users_events
    - Effect: Allow
      Action: dynamodb:*
      Resource: arn:aws:dynamodb:eu-central-1:178190218027:table/EU_users_events
    - Effect: Allow
      Action: s3:*
      Resource: arn:aws:s3:::sfn-blog
plugins:
  - serverless-step-functions
# Define a path to the AWS Lambda Functions
functions:
  FetchFromDynamoDBState:
    handler: handler.FetchFromDynamoDBState
  ProcessUSUsersEvents:
    handler: handler.ProcessUSUsersEvents
  ProcessEUUsersEvents:
    handler: handler.ProcessEUUsersEvents
  InsertMessageDynamoDBTest:
    handler: handler.InsertMessageDynamoDBTest
  VerifyStepFunctionOutcomeTest:
    handler: handler.VerifyStepFunctionOutcomeTest
# Define Step Function and its state machine
stepFunctions:
  stateMachines:
    proceedRewards:
      name: proceedRewards
      definition:
        StartAt: InsertMessageDynamoDBTest
        States:
          InsertMessageDynamoDBTest:
            Type: Task
            Resource:
              Fn::GetAtt: [InsertMessageDynamoDBTest, Arn]
            ResultPath: "$.response"
            Next: WaitState
          WaitState:
            Type: Wait
            Seconds: 200
            Next: FetchFromDynamoDBState
          FetchFromDynamoDBState:
            Type: Task
            Resource:
              Fn::GetAtt: [FetchFromDynamoDBState, Arn]
            ResultPath: "$.items"
            Next: ProcessDataState
          ProcessDataState:
            Type: Map
            ItemsPath: "$.items.items"
            ResultPath: "$.mappedData"
            MaxConcurrency: 2
            Iterator: 
              StartAt: ProceedChoiceState
              States:
                ProceedChoiceState:
                  Type: Choice
                  Choices:
                    - Variable: $.region.S
                      StringEquals: "US"
                      Next: ProcessUSUsersEventsState
                    - Variable: $.region.S
                      StringEquals: "EU"
                      Next: ProcessEUUsersEventsState
                  Default: DefaultState
                ProcessUSUsersEventsState:
                  Type: Task
                  Resource:
                    Fn::GetAtt: [ProcessUSUsersEvents, Arn]
                  End: true
                ProcessEUUsersEventsState:
                  Type: Task
                  Resource:
                    Fn::GetAtt: [ProcessEUUsersEvents, Arn]
                  End: true
                DefaultState:
                  Type: Fail
                  Cause: 'Invalid region value.'
                  Error: 'InvalidRegionError'
            Next: VerifyStepFunctionOutcomeTest
          VerifyStepFunctionOutcomeTest:
            Type: Task
            Resource:
              Fn::GetAtt: [VerifyStepFunctionOutcomeTest, Arn]
            End: true

The additional Lambdas for triggering GitHub workflows should look like this:

const axios = require('axios');
module.exports.InsertMessageDynamoDBTest = async () => {
await triggerWorkflow(process.env.INSERT_ACTION_TRIGGER_ID);
}
module.exports.VerifyStepFunctionOutcomeTest = async () => {
await triggerWorkflow(process.env.VERIFY_ACTION_TRIGGER_ID);
}
const triggerWorkflow = async (workFlowId) => {
const owner = process.env.GITHUB_OWNER;
const repo = process.env.GITHUB_REPO;
const token = process.env.GITHUB_TOKEN;
let response;
try {
        response = await axios.post(
`https://api.github.com/repos/${owner}/${repo}/actions/workflows/${workFlowId}/dispatches`,
            {
                ref: 'main'
            },
            {
                headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${token}`
                }
            }
        );
console.log('Job triggered successfully:', response.data);
    } catch (error) {
console.error('Failed to trigger job:', error.response.data);
    }
return response;
}

The final version of the Step Function should look like the following:

This way, we will ensure that our Step Function continuously processes the messages successfully.

Note: The testing states should be present only in the testing environments.

Alternatives

We could implement test automation for this kind of system by using other approaches. For example, we could create scheduled triggers for GitHub Actions Workflows. The first one for data insertion to the DynamoDB could be scheduled at the beginning of the day, and the second one for the verification of processed data could be scheduled at a time when we expect that at least one Step Function Workflow execution has been finished. The advantage of this approach is that we could avoid using the first and the last lambda functions for triggering the GitHub Actions Workflows that we specified in our Test Automation part. The advantage of the first approach that we explained is that if someone makes some changes on the Step Function, it will be verified on the next execution. We could also implement both approaches with Jenkins and any other CI/CD tool.

The main challenges

The main challenges for the implementation of test automation on systems based on Step Functions (systems that run periodically or per some event) are unique, and here are some of the main ones:

1. Time-dependent testing: Systems triggered periodically often involve time-based operations. We may need to control the passage of time during tests to validate time-dependent behaviors.
2. Data Security: We must ensure that test data are handled securely and that the test environments are adequately isolated, especially if we are working with sensitive data.
3. End-to-End Testing: We need to ensure that all components work together, and it often requires two or more separate scripts that will orchestrate together by exchanging the data as artifacts for testing just one system behavior.
4. Continuous Integration/Continuous Deployment (CI/CD): We must ensure that tests run automatically with each deployment.
5. Handling Large Data Sets: Systems with Step Functions may process large data sets. Testing with large data can take time and resources, so it needs a particular approach.
6. Environment Consistency: To catch environment-specific issues, test environments should closely resemble production environments, including Cloud configurations.
7. Resource Setup and Teardown: Setting up and tearing down AWS resources, such as databases, queues, or S3 buckets, for each test can be time-consuming and costly.
8. Data Generation: Creating test data to simulate real-world scenarios and edge cases can be challenging. We must ensure that the data is consistent with the state transitions in our Step Function workflows.
9. Connection with External Services: Establishing connections to the different services like databases and etc., can sometimes be very challenging based on many different situations, including security configurations, etc.

Conclusion

We have gone through the overall concept of development and application of the Step Function and the integration of automation testing within its workflow. Through all this, we were introduced to additional tools from the domain of cloud computing, CI/CD, and test automation.

You can find the full project on this GitHub repository.

Sources:

https://docs.aws.amazon.com/step-functions/latest/dg/concepts-states.html

If you found this useful, check out other Atlantbh blogs!

Originally published at https://www.atlantbh.com on November 20, 2023.

Test Automation Integrated into AWS Step Function Workflow was originally published in Atlantbh Engineering on Medium, where people are continuing the conversation by highlighting and responding to this story.

Uniting different skill sets, experiences, and perspectives through engineering collaboration is a process that brings complex challenges, but also can forge innovation through synergy if done correctly. Atlantbh had a chance to work on such a project where close day-to-day collaboration with an external development team was needed for successful delivery. In this case study, we’ll explore how we collaborated to update the system’s architecture, handle changes in technology and onboard to Elixir, and navigate the business landscape while nurturing a partnership built on adaptability and shared knowledge.

The problem

The client had been developing a backend system for a number of years, for which they realized that the initial architecture and application model were no longer suitable to their current needs. As the business was spreading, this implicated a rise in the number of end-users and different use cases which haven’t been foreseen and didn’t fit in the current architecture. To bypass system limitations, numerous temporary solutions needed to be implemented to address these gaps. Unfortunately, when the system outgrows its initial architecture and temporary solutions come into play for a longer period, at some point the development team finds itself fully consumed with fixing production issues that arise due to this very situation, as was the case with this particular project.

Consequently, a decision has been made to rewrite the old backend system using a more appropriate tech stack and to create a flexible model capable of accommodating all use cases. In search of a partner, the client was looking for someone who would invest considerable time and expertise in analyzing the requirements, learning both the business and new technology, collaborating closely with the internal technical team, and successfully integrating into the existing ecosystem.

Challenges

During the initial weeks, we devoted time to gathering and analyzing requirements. Additionally, we have invested efforts in familiarizing ourselves with the legacy application, and the whole ecosystem into which our service would integrate, addressing technical challenges, and gaining an understanding of the client’s expectations.

The team was faced with the following challenges:

• New technology stack — Elixir. The transition from Java (our usual choice of technology) to Elixir required learning and adapting to the new technology and frameworks. We did have expertise in Ruby which is a similar programming language, but not with the particular one. What also helped is the fact that our software engineers are full-stack engineers, and are used to adopting different technology stacks.
• Event-driven model approach. Adopting an event-driven model for software development introduced a paradigm shift in our development approach. It already helped that we had expertise in event-based communication, designing asynchronous workflows, etc.
• Complex ecosystem and business logic. Our software was part of a larger ecosystem, including multiple interconnected services that shared data and events. We did have experience in working with different teams in large ecosystems, but this one was quite different in size and complexity. We had to find the best collaboration model which would work for both teams, to understand dependencies and common practices.

Approach

Our journey with Elixir

The client offered assistance in knowledge transfer and onboarding to the new technology stack to address these challenges. We conducted knowledge transfer and pair programming sessions, sharing expertise and guiding us to avoid common pitfalls when starting with this stack. The team was grateful for such an opportunity, which leveraged our efforts later. We continued independently familiarizing ourselves with Elixir, and started the development of standalone features.

Collaboration model

The chosen collaboration model involved regular communication and coordination between the teams:

• Daily sync calls were conducted, with representatives from both teams sharing updates and discussing progress.
• Weekly sync meetings between the product manager, technical lead, and our service lead helped align expectations and address any concerns.
• Planning and breakdown sessions were organized as needed to ensure efficient project management and decision-making.
• All parties were open to constructive feedback, which helped us overcome challenging situations and manage expectations.

Development and delivery

In collaboration with a product manager and technical lead, we defined the scope for the initial version of the product, ensuring it would provide immediate value to the users. Our team was responsible for the backend service which would replace a specific segment of the old system.

As the business logic defined was unlikely to undergo significant changes in the future, we have agreed on the approach that involved implementing the defined functionalities one by one, in detail, incorporating comprehensive logic and thorough unit and integration testing. To document the system behavior, we adopted the Gherkin language which was already being used in the rest of the ecosystem, ensuring clarity and future stability. While this approach demanded more time for development and delivery upfront, the client was aware of the long-term benefits of investing in a solid foundation.

As of now, our service has been successfully deployed in production, catering to a small user base. By diligently gathering feedback and continuously adapting to improve our service, we are moving on our way to transition to larger user groups. The feedback received thus far validated the stability and reliability of our core logic.

Conclusion

Each new project presents its unique challenges, but a willingness to embrace new approaches and technologies brings significant advantages to software companies, broadening their horizons and knowledge base.

Throughout this endeavor, our team demonstrated key characteristics that played an essential role in a successful partnership with our client. Our expertise in software delivery, profound technical knowledge of design principles and software engineering, problem-solving skills, and readiness to embrace new methodologies and feedback were instrumental in achieving positive outcomes.

Technology stack: Elixir, Phoenix, HTML, CSS, Rabbitmq, Gherkin
Project duration: 1+ year

Blog by Irma Smajić, Product Owner at Atlantbh

Crafting Collaboration in Elixir: A Case Study of Success in Software Development was originally published in Atlantbh Engineering on Medium, where people are continuing the conversation by highlighting and responding to this story.

Lately, we have seen a significant number of discussions regarding the potential of AI and strategies to incorporate it into the product development process. Specifically, how can we enhance, innovate, and scale in ways we were not able to do before? In this interview, we asked Samra Tanovic, our VP of Analytics and Services, to share her experience and insights on this topic.

Q: Could you tell us about your journey in incorporating AI into products?

A: Several years ago, when the need for AI functionalities arose within one of our projects, we were faced with a series of questions: How do we enable our teams and build the expertise needed for such tasks? How long will it take? How do we spot opportunities for the use of AI faster? What are the costs?

The key takeaway from this experience is that it’s essential to start with an explicit, well-defined goal that aligns with business and user needs and then to consider how AI could assist in achieving this goal. It’s important to recognize that not all problems require an AI solution. There’s a risk of wasting significant time and resources trying to utilize AI just for the sake of using it. Instead, we learned to carefully examine existing challenges and ask: Can AI solve this problem more efficiently? Is there potential for scalability? Does the benefit of using AI outweigh the cost?

Our journey also taught us that data science projects, unlike standard software engineering, involve a series of experiments that may or may not lead to discoveries or entirely new lines of research. The inherent uncertainties in this process make it difficult to define and control, which is why a unique management approach is required.

Another key aspect we realized is the importance of the data that AI uses and produces. While the size of the training data does impact the model’s performance, data quality often plays a considerably more influential role. Investing in clean, relevant, and diverse data preprocessing turned out to be essential for accurate AI outcomes, making it an investment worth prioritizing.

Finally, in order to build expertise and prepare our company for the rapidly evolving AI landscape, we set up a team dedicated to the research, development, and implementation of cutting-edge AI solutions within existing and new projects. This step allowed us to continue innovating and incorporating AI where it made sense, enhancing both our products and our ability to adapt to future technological advancements.

Q: Could you provide examples where your teams used AI to enhance problem-solving efficiency?

A: The main criterion is to consider processes that are excessively manual, do not scale well with product growth, or areas where customer experiences could be improved through personalization or automation. ML is mostly required in cases when the targeted behavior cannot be accurately expressed in software logic without relying on external data sources.

A few years back, we used AI to analyze text and extract common business attributes, such as business category, address and operating hours from an unstructured text, a process that was previously time-consuming for humans. The efficiency of the validation of the business attributes, which was our main use case, was increased by 80%.

An interesting and simple example involved removing a common UI element, a dropdown menu, and using classification to determine the correct option automatically. Previously, selecting the wrong option led to downstream problems and complicated user experience. However, by leveraging ML to accurately determine the appropriate dropdown option based on the user’s input, we managed to reduce mistakes by 30%.

Moreover, an intriguing area to consider is threshold replacements. Traditional business rules with set thresholds are often used to trigger specific actions. However, these thresholds can be too rigid and may not adapt well to changing circumstances. AI models, in contrast, can learn from data patterns and make more flexible, dynamic decisions.

Q: Lastly, drawing from your experience, what are some obstacles that businesses might encounter while implementing AI, and what strategies can they employ to overcome them?

A: One of the biggest challenges is the lack of trust in AI-powered solutions. Many businesses are hesitant to rely on AI for critical decisions because they don’t fully understand how algorithms work or the resulting accuracy is not meeting their expectations.

AI systems are designed to learn but can make errors or provide inaccurate results based on various factors, such as the quality of the training data, complexity of the task, and limitations of the algorithms used.

In scenarios where explicit and deterministic results are critical, businesses should carefully assess the suitability of AI models. Hybrid approaches that pair AI with deterministic methods should be considered, along with models with high interpretability, to maintain control and build trust. It is also useful to be transparent about limitations and potential biases.

Speaking of biases, if not designed and trained with fairness and ethical considerations in mind, AI systems have the potential to perpetuate biases and discrimination. Ensuring compliance with regulations, regular audits, testing for bias and seeking legal advice can help.

Q: Thank you. It was a pleasure discussing this topic.

A: Thank you. I’m excited to see how AI continues to shape and enhance the field of product development.

Originally published at https://www.atlantbh.com on July 24, 2023.

Unleashing the Power of AI in Product Development was originally published in Atlantbh Engineering on Medium, where people are continuing the conversation by highlighting and responding to this story.

Data analysis often presents us with the challenge of dealing with noisy inputs. This is particularly evident when working with large datasets of user inputs. Detecting and filtering out random string inputs can prove invaluable in various scenarios, such as data validation, quality control or the development of machine learning (ML) based natural language processing (NLP) systems (check for data quality improvement steps in this blog post).

What should/can be considered a ‘random string’?

The term “random string” may seem straightforward at first glance, but its interpretation can be subjective and context-dependent. What one person perceives as a random string, another may see as a meaningful pattern or representation. This relativity is particularly evident when considering visual or linguistic aspects.

Let’s delve into a few examples to highlight the importance of relativity when defining random strings. Take, for instance, the string ‘\__o__/’. At first glance, it might appear random, but one could argue that it resembles a stick figure of a man with raised hands. In this case, the interpretation of randomness becomes subjective, as different individuals may perceive different patterns or associations.

Similarly, let’s consider the word “Szczesny.” In the English language, it may appear as a random combination of letters, devoid of any meaning. However, in Polish, it happens to be the surname of a famous goalkeeper (Image 2). This example demonstrates how the interpretation of randomness can be influenced by language and cultural context. What may seem random in one language could hold significance or meaning in another.

It’s crucial to acknowledge that randomness is not an inherent property of a string itself but rather a perception based on patterns or associations that individuals may perceive. In the context of data analysis and ML, understanding this relativity is vital when developing algorithms or models that aim to detect and filter out random strings.

When designing systems to identify random strings, it becomes essential to consider the specific domain, context, and purpose of the analysis. What may be considered random noise in one scenario could be a valuable signal in another. For instance, when analyzing social media data, a sequence of random characters might indicate spam or bot-generated content. On the other hand, in cryptography, random strings play a crucial role in generating secure encryption keys.

Moreover, the concept of randomness itself can vary depending on the intended application. Some algorithms generate strings that may appear random to human observers but possess specific statistical properties that make them suitable for cryptographic purposes. In contrast, other applications require strings that exhibit specific patterns or structures to serve their intended function.

How to detect a random string of characters in a specific locale?

If the goal is to detect random strings which consist of a specific set of characters (e.g., English alphabet characters), one possible way of detecting random strings is using bigrams.

To be able to do that, you need to create a reference dictionary of bigrams for the considered locale. For the English language, we can use bigrams and their frequencies from Peter Norvig’s analysis. For example, in English language text, the bigram “th” is very common, while the bigram “zx” is not. Using the available list of bigrams and their frequencies, it is possible to normalize it (e.g., min-max normalization) and create a reference dictionary. An important note here is that you can easily create your own list of bigrams (extracted from word corpus) for any locale if you are able to find open-source books, newspapers or magazines.

Reference dictionary is a dictionary of all 26² = 676 combinations of English bigrams with associated frequency numbers (from 0 to 100) where higher values represent more frequent bigrams (like “th”) and lower values represent less frequent bigrams (like “zx”). Users are able to adjust the threshold value (default value is 0.1) in order to define a boundary between less common and common bigrams. For example, if the threshold is 0, all bigrams will be considered as common bigrams. Otherwise, if the threshold is 100, all bigrams will be considered as less common bigrams. You can find a reference dictionary of bigrams here.

To consider whether a word is a random string or not, here is a list of steps that are used to help detect random strings:

1. Length and Character Validation: The first checkpoint of the algorithm verifies that the word is longer than three characters (words up to three characters may be regular abbreviations) and consists solely of English alphabetic characters. This ensures that we are working with meaningful words rather than arbitrary combinations of symbols
2. Repeating Characters: If the entire word is made up of the same character, such as “hhhhhhh”, it is immediately labeled as random. This is because the presence of repeating characters indicates a lack of complexity and structure, suggesting a random generation
3. Lowercasing the Word: To eliminate the case sensitivity, the algorithm converts the word to lowercase. This prevents the distinction between uppercase and lowercase characters from affecting the detection process
4. Bigrams Analysis: A bigram refers to a sequence of two consecutive characters in a word. The algorithm generates a list of all possible bigrams within the word. For example, the word “random” would produce the following bigrams: ra, an, nd, do, om
5. Common and Uncommon Bigrams: Each bigram is compared against a reference dictionary of common English bigrams. If a bigram has a frequency above a given threshold, it is considered common; otherwise, it is classified as uncommon. The choice of threshold depends on the desired sensitivity of the algorithm
6. Counting Common and Uncommon Bigrams: The algorithm calculates the number of common and uncommon bigrams in the word. If the count of common bigrams exceeds that of uncommon bigrams, the word is classified as non-random. Conversely, if the count of uncommon bigrams surpasses that of common bigrams, the word is deemed random

Here are Python and Java codes for the proposed algorithm:

Edit: Check for the updated version of random-string-detector at https://github.com/mehmedkadric/random-string-detector.

Python:

def is_random_string(word, threshold=0.1):
    # Allow only words longer than 3 characters which contains only English alphabetic characters
    if len(word) < 4 or not word.isalpha():
        return False

    # Repeating characters
    if len(set(word)) == 1:
        return True

    # Turn word into lowercase
    word = word.lower()

    # Get list of bigrams from the word
    bigrams = [word[i:i + 2] for i in range(len(word) - 1)]

    # Get number of common and uncommon bigrams
    num_common_bigrams = sum(1 for bigram in bigrams if en_bigrams_dict.get(bigram, 0) > threshold)
    num_uncommon_bigrams = len(bigrams) - num_common_bigrams

    # Higher number wins
    if num_common_bigrams > num_uncommon_bigrams:
        return False
    else:
        return True

Java:

package checker;


import java.util.HashMap;
import java.util.Map;


public class RandomStringChecker {
   private static final double DEFAULT_THRESHOLD = 0.1;
   private static final Map<String, Double> enBigramsDict = new HashMap<>();


   static {
       // Populate the dictionary of English bigrams here
       enBigramsDict.put("ab", 6.461565670356169);
       enBigramsDict.put("bc", 0.0531330714265234);
       enBigramsDict.put("cd", 0.06273467822837461);
       // ...
   }


   public static boolean isRandomString(String word) {
       return isRandomString(word, DEFAULT_THRESHOLD);
   }


   public static boolean isRandomString(String word, double threshold) {
       // Allow only words longer than 3 characters which contain only English alphabetic characters
       if (word.length() < 4 || !word.matches("[a-zA-Z]+")) {
           return false;
       }


       // Repeating characters
       if (word.chars().distinct().count() == 1) {
           return true;
       }


       // Turn word into lowercase
       word = word.toLowerCase();


       // Get list of bigrams from the word
       String[] bigrams = new String[word.length() - 1];
       for (int i = 0; i < word.length() - 1; i++) {
           bigrams[i] = word.substring(i, i + 2);
       }


       // Get number of common and uncommon bigrams
       int numCommonBigrams = 0;
       for (String bigram : bigrams) {
           if (enBigramsDict.containsKey(bigram) && enBigramsDict.get(bigram) > threshold) {
               numCommonBigrams++;
           }
       }
       int numUncommonBigrams = bigrams.length - numCommonBigrams;


       // Higher number wins
       return numCommonBigrams <= numUncommonBigrams;
   }
}

This algorithm has been released as a Python package and can easily be installed using the command:

pip install random-string-detector

Examples

Python method calls:

from random_string_detector.random_string_detector import is_random_string

print(is_random_string('abcd')) # True
print(is_random_string('Thgrbh')) # False
print(is_random_string('Thgrbh', 5)) # True - threshold adjusted

Java method call:

public static void main(String[] args) {
   System.out.println(isRandomString("abcd")); // true
}

Conclusion

The interpretation of randomness depends on the specific domain, context, and purpose of the analysis. What may be considered random noise in one scenario could be valuable information or a critical signal in another. It is essential to align the definition of random strings with the objectives and requirements of the particular application to ensure the reliability and quality of results.

While the proposed algorithm is not foolproof and cannot definitively determine randomness, it serves as a useful tool for various applications that require distinguishing between random and non-random strings.

A potential algorithm improvement would be to have the bigram dictionary in a separate file that we could load into memory.

Appendix

PyPI: https://pypi.org/project/random-string-detector/

GitHub: https://github.com/mehmedkadric/random-string-detector

Random String Detection was originally published in Atlantbh Engineering on Medium, where people are continuing the conversation by highlighting and responding to this story.

We can agree that AI is not a one-size-fits-all solution. However, for many companies, AI can provide significant benefits and help drive growth and success. Some potential benefits include the following:

• Improved efficiency: AI can automate tasks and processes, saving time and resources for other activities.
• Increased accuracy: AI can make more accurate predictions and decisions than humans, reducing the risk of errors.
• Enhanced customer experience: AI can personalize customer interactions, providing a more tailored and seamless experience.
• Cost savings: AI can help businesses reduce costs by automating tasks, improving efficiency, and reducing the need for human labor.
• Competitive advantage: Companies that use AI can gain a competitive advantage over those that don’t by being able to analyze and act on data faster and more effectively.

The last decade has brought significant growth in the field of Data Science in general. However, the total value of AI is still locked in many sectors, such as health care, manufacturing, and government technology. In a study published by Accenture, 80% of all Proof of Concepts (PoCs) do not make it into production. What exactly is going wrong, and why is AI not as successful in the real world as it is in academic studies?

Model-centric vs. Data-centric approach

A strong model-centric AI (traditionally presented in academia) considers data only as a static parameter. If a model’s performance is not as expected, AI/ML engineers will try to tune the model’s hyperparameters or even change the model while data stays untouched.

Subject Matter Experts do not play a vital role in the whole process of developing AI-based systems in a model-centric approach. Instead, AI/ML engineers usually make business-related decisions during that process.

As per Andrew Ng, an AI system is a combination of code and data. While a significant amount of time and resources have been dedicated to developing code and algorithms, it is now necessary to prioritize improving data quality and relevance to achieve desired outcomes.

By shifting the focus to data, we are basically moving from model-centric to data-centric AI.

In a data-centric approach, the role of Subject Matter Experts is significant, and AI systems are much closer to every business.

Adopting a data-centric approach in the real world or during work on a practical project is more fruitful. That means the model stays fixed and focuses on feeding the model with good or high-quality data. A decent algorithm with good data may even outperform a great algorithm with not-so-good data.

Data should not be the only priority in the development process. It is also important to consider the structure and functionality of the model and code used. The choice of model can significantly impact the accuracy, as demonstrated in the analysis of the Titanic disaster, where the prediction accuracy varied from 78% to 94%, depending on the algorithm.

Big Data vs. Good Data: What’s More Important for AI?

Good Data and Big Data are two terms often used in the field of Data Science and AI. Good data refers to data that is accurate, relevant, and well-organized. Big data, on the other hand, refers to large datasets that are too large and complex to be processed and analyzed using traditional data processing tools and techniques.

In order for AI algorithms to make accurate predictions and decisions, they need to be trained on large amounts of data. However, simply having a lot of data is not enough. The data must also be high-quality and well-organized for the AI model to learn effectively.

In many cases, companies only have access to small datasets, which can lead to poor results if the focus is solely on the model. Andrew Ng emphasizes the benefits of a data-centric approach to machine learning and suggests that there should be a shift towards this approach within the community. He uses the example of a steel defect detection problem, in which the data-centric approach improved the model’s accuracy by 16% compared to a model-centric approach [Source: neptune.ai].

Data Quality Issues

Many famous datasets have data quality issues. For instance, The COCO and ImageNet datasets are widely used in the field of computer vision for tasks such as object detection, image classification, and image segmentation. While these datasets are widely used and generally considered high quality, some issues have been discovered with them that may impact their usefulness for specific tasks. Some of the problems detected in the COCO dataset include mislabelled objects (first image) and labeling inconsistencies (second image) [Source: neuralception.com].

Ensuring Data Quality in Machine Learning Development

While the specific methods for improving data quality may vary depending on a specific problem, some general steps can be applied in many situations:

• Verify data sources.
• Ensure that data sources are reliable and accurate.
• Make sure your model gets enough data to be able to generalize.
• Validate and clean data.
• Check labeling inconsistencies.
• Perform validation checks and look-ups on labeled data if applicable.
• Check labels distribution and manually check entries with low-frequency labels.
• Try to make a tool that will give you suspicious labels so you can check them manually.
• Standardize data to a common format or schema.
• Involve Subject Matter Experts.
• Document agreements after discussing inconsistencies with labelers.
• Analyze data and perform feature engineering.
• Train a model with labeled data and apply the model to training data. Check the diffs.
• Perform error analysis.

The development of machine learning-based systems is a highly iterative process. Keeping track of your notes, findings, and conventions in a documentation system can help maintain data quality. By following mentioned steps and being vigilant about potential issues, you can help ensure that your data is of high quality and ready for use in your machine-learning projects.

Wrapping Up

It is important to consider the quality and quantity of data and the structure and functionality of the model and code used. By prioritizing data quality and involving subject matter experts, companies can improve the accuracy and effectiveness of their AI systems, drive growth and success, and gain a competitive advantage.

Blog by Mehmed Kadrić, Senior Data Analyst at Atlantbh

Originally published at https://www.atlantbh.com on January 10, 2023.

Data-centric AI / Big data vs. Good data was originally published in Atlantbh Engineering on Medium, where people are continuing the conversation by highlighting and responding to this story.