Apache Kafka News

• 

  3 Ways to Prepare for Disaster Recovery in Multi-Datacenter Apache Kafka Deployments
  cloud confluent platform replicator disaster recovery multi-datacenter replicat white paper • • sorangutan  

  1
  1
  Posts
  112
  Views

  

  Imagine:

  Disaster strikes—catastrophic hardware failure, software failure, power outage, denial of service attack or some other event causes one datacenter with an Apache Kafka^® cluster to completely fail. Yet Kafka continues running in another datacenter, and it already has a copy of the data from the original datacenter, replicated to and from the same topic names. Client applications switch from the failed cluster to the running cluster and automatically resume data consumption in the new datacenter based on where it left off in the original datacenter. The business has minimized downtime and data loss resulting from the disaster, and continues to run its mission critical applications.

  Ultimately, enabling the business to continue running is what disaster recovery planning is all about, as datacenter downtime and data loss can result in businesses losing revenue or entirely halting operations. To minimize the downtime and data loss resulting from a disaster, enterprises should create business continuity plans and disaster recovery strategies.

  There are three actions you should take today for disaster planning:

  1. Design a multi-datacenter solution
  2. Build a runbook for failover and failback
  3. Test! Test! Test!

  1. Design a multi-datacenter solution

  A Kafka deployment in a single datacenter provides message durability through intra-cluster data replication. Data replication with producer setting acks=all provides the strongest available guarantees, because it ensures that other brokers in the cluster acknowledge receiving the data before the leader broker responds to the producer.

  The single datacenter design provides robust protection against broker failure. If a client application is using a certain broker for connectivity to the cluster, and that broker fails, another bootstrap broker can provide connectivity to the cluster. However, this single Kafka deployment is vulnerable if the entire datacenter fails.

  In a multi-datacenter design, instead of a single datacenter with one Apache Kafka deployment, there are two or more datacenters with Kafka deployments geographically dispersed. They can be on prem or in Confluent Cloud. Confluent Replicator synchronizes data between the sites so that in the event of a partial or complete disaster in one datacenter, applications can fail over to the other datacenter.

  To help you design your multi-datacenter solution, we’ve released an updated version of the white paper Disaster Recovery for Multi-Datacenter Apache Kafka Deployments. The latest version of the paper walks through the design and configuration of a multi-datacenter deployment, covering designs, centralized schema management and client application development for multi-datacenter.

  It also discusses in detail the new Confluent Replicator features in version 5.0, including:

  • Prevention of cyclic replication of topics: In the context of active-active datacenters with Replicator running in both directions, Replicator can prevent infinite loops. Otherwise, if you don’t use special topic naming patterns, copying topics could result in an infinite loop copying data from cluster-1 to cluster-2, and then copying that same data in cluster-2 back to cluster-1 and then again from cluster-1 to cluster-2.
  • Timestamp preservation: The time that is recorded at initial message production in the origin cluster propagates to the destination cluster. This enables streaming applications to act on the original timestamps.
  • Consumer offset translation: The committed consumer offsets are replicated to the destination cluster and translated in such a manner that the offset values make sense in that cluster. This allows consumers to switch to the destination cluster and automatically resume consuming messages where they left off.

  Your architecture will vary depending on your business requirements, but you can apply the building blocks from this white paper to strengthen your disaster recovery plan.

  

  2. Build a runbook for failover and failback

  When disaster strikes, you have to be prepared with a runbook of specific actions to take. How do you fail over client applications to the new datacenter? How do you enable Confluent Schema Registry to register new schemas? What do you have to do to enable consumers to start resuming data consumption in the new cluster? Most importantly, after the original datacenter recovers, how do you failback?

  Disaster Recovery for Multi-Datacenter Apache Kafka Deployments walks through the basic multi-datacenter principles, explaining what to do if one datacenter fails and how to failback when it recovers. You should adapt those principles to your specific deployments.

  How manual or automated the failover workflow is depends on the recovery time objective (RTO), which is the point in time after the disaster event when the failover completes. Ideally, it is as soon as possible after the disaster to minimize downtime. The more aggressive the RTO is, the more you want an automated workflow, which depends on the service discovery mechanism and failover logic developed into the applications.

  The recovery point objective (RPO) is the last point in time before the disaster event that the application can recover to with known good data. Usually it is as close before the disaster as possible to minimize data loss. Especially if you leverage the features provided in Replicator 5.0, it will immensely simplify the workflow for failover and failback, as well as improve these recovery times.

  3. Test! Test! Test!

  Use the building blocks in the updated white paper to configure an environment that is representative of your production environment, then simulate the disaster. Follow the entire workflow: Take down a cluster, fail over applications, make sure they are successfully processing data and then fail back.

  Stepping through the entire workflow is important primarily because it ensures that the multi-datacenter design is robust and working for your specific business needs. Every deployment is slightly different, and client applications are written slightly differently to meet their business needs, so you need to make sure that everything is functional. The second reason to do this simulation is it inherently tests your runbook to ensure that the steps are well documented.

  To help you with the testing, Confluent provides a Docker setup in a GitHub repository that you can clone. Disclaimer: This is just for testing—do not take this Docker setup into production!

  The Docker brings up an active-active multi-datacenter environment leveraging the new Replicator features, such as provenance headers to prevent cyclic replication of topics, timestamp preservation and consumer offset translation.

  For testing, adapt the configurations to be more representative of your deployment and run your client applications against it. You may also use the provided sample client application to see how data consumption can resume in the new datacenter based on where it left off in the original datacenter. Then, simulate the failover and failback, test its behavior, test your runbook and verify that it all works.

  

  In summary, a disaster recovery plan often requires multi-datacenter Kafka deployments where datacenters are geographically dispersed. Take these three steps to ensure you are prepared:

  1. Design a multi-datacenter solution
  2. Build a runbook for failover and failback
  3. Test! Test! Test!

  You may be considering an active-passive design (one-way data replication between Kafka clusters); active-active design (two-way data replication between Kafka clusters); client applications that read from just their local cluster or both local and remote clusters; service discovery mechanisms to enable automated failovers; geo locality offerings; etc.

  The latest version of the white paper Disaster Recovery for Multi-Datacenter Apache Kafka Deployments provides building blocks that you can apply to any of the above.

  
  
  https://www.confluent.io/blog/3-ways-prepare-disaster-recovery-multi-datacenter-apache-kafka-deployments
• 

  Streaming in the Clouds: Where to Start
  cloud multi-cloud hybrid cloud confluent cloud replicator cloud first • • sorangutan  

  1
  1
  Posts
  111
  Views

  

  Only a few years ago, when someone said they had a “cloud-first strategy,” you knew exactly who their new preferred vendor was. These days, however, the story is a lot cloudier.

  Every conversation around cloud now involves terms like hybrid cloud and multi-cloud. Executives want to avoid vendor lock-in and operate as close to their data sources as possible. Developers and data scientists would like the flexibility to leverage best-in-class services for their specific use case from different cloud providers. A strategy focused around a single, massive cloud environment just doesn’t cut it in many cases.

  However, these complex cloud strategies create new challenges. In an ideal situation, developers build cloud-native applications, leverage cloud ecosystem services and move seamlessly between cloud providers without having to worry about where their data is. But this is difficult.

  Companies are just getting a grip on how to move some of their data to the cloud and build new applications specifically in and for the cloud. It’s a massive challenge to forklift existing systems of record and associated applications that have been the backbone of a company’s operations into cloud. How can they possibly architect for moving data between multiple applications and clouds, and having the relevant data show up across environments in real time, regardless of where it was created?

  So even if companies are looking at cloud-first as their ultimate goal, many end up working in a hybrid cloud model in the short to medium term where some applications run on-premises and some run in the cloud. Imagine the challenges in the hybrid cloud world of connecting legacy systems with new cloud applications and making it appear as a single cohesive system across the company.

  A common anti-pattern we see looks like the diagram shown below—HTTP requests across the WAN which creates a point-to-point architecture. It works initially but has several drawbacks. It makes for a slow, weak bridge with massive interconnections, with different systems talking to the cloud services in different ways. This is not a scalable architecture.

  

  Leveraging Apache Kafka for cloud migration and beyond

  Fortunately, Apache Kafka^® has risen to prominence as a key enabler of hybrid, cloud-only and multi-cloud strategies. Kafka has been the leader among technologies used in developing streaming data pipelines and apps for several years. As it turns out, organizations are finding that many of the components that make Kafka great for building and managing real-time pipelines and apps are also incredibly valuable in building a bridge to cloud.

  Alternatively, Apache Kafka can also be used to build data pipelines across cloud regions to address disaster recovery (DR) requirements that your organization might have. This can be a more practical way to implement DR strategies instead of relying on costly proprietary solutions.

  For example, let’s consider Confluent Replicator, which actively replicates data across datacenters and public clouds. It provides a simple and scalable solution to bridge your data into the cloud. Below are some examples of how Confluent Replicator can help you build a bridge to cloud, or even a bridge between clouds.

  Hybrid Cloud (environment spanning on prem and cloud)	Multi-Cloud (environment spanning different cloud providers)	Cloud Only (environment only in the cloud)
  Replicate data between on-prem datacenter and public clouds continuously and reliably.	Replicate data between public clouds.	Replicate data between clouds, across applications or anywhere else.

   

  So let’s take a look at how you might build a hybrid architecture that allows for a step-by-step migration. Instead of having many point-to-point connections, you can build a hub and spoke implementation based on a central Kafka platform.

  You can publish streams of data from your local on-prem environment, replicate it to different cloud regions and environments, load into data systems in the cloud and trigger cloud-native applications off of those events and data. So the data stays in sync across both on prem and cloud. It shows up across the infra, in near real time, and what you end up with is a future-proof, massively scalable platform to support all of your company’s needs.

  

  Once you have your data in the cloud, you can take advantage of the broad Apache Kafka ecosystem of connectors to easily and quickly scale out to various cloud services and data systems. The most popular data systems, such as S3 on AWS, Elasticsearch, GCS on GCP, etc., have prebuilt connectors from Confluent, its partners or the Kafka community.

  Comparing Confluent Replicator and MirrorMaker

  While Confluent Replicator has been compared to open source MirrorMaker in the past, there are enterprise-critical features in Confluent Replicator that cannot be found in MirrorMaker. Yes, both technologies provide data, schema and Connect replication, which are all extremely important in managing data across datacenters and clouds. But MirrorMaker only partially supports flexible topic selection and auto-creation of topics.

  And, there are loads of enterprise-grade features in Confluent Replicator that are not supported in MirrorMaker. Confluent Replicator automatically detects and replicates new partitions in your streams. It replicates topic configuration between two clusters so your two clusters are always in sync. With Confluent Replicator, you can filter, modify and route events on the fly. (For a thorough comparison, please see Confluent documentation on multi-datacenter replication.)

  Given all of the above, it makes a lot of sense to leverage Apache Kafka for your cloud migration. It means setting up Kafka clusters in the source and destination environments, sizing the clusters for the right throughput, ensuring availability, addressing security requirements and all of the other nitty gritty operational details that go into building distributed systems.

  You’re thinking all the benefits sound great but implementation takes time, expertise and management. How might you speed up your migration or Kafka deployment, and offload some of the management burden while ensuring peace of mind and reliability?

  Would a fully managed Kafka service help?

  Confluent Cloud provides Apache Kafka as a fully managed service in your cloud of choice with 99.95 percent uptime SLA. It guarantees under one-hour response time for high priority issues from Kafka experts—the same people who originally created Kafka and commit regularly to Kafka source code. So why not offload the burden of deploying, upgrading and maintaining Kafka to the experts, and instead focus on your migration strategy and plan?

  
  
  https://www.confluent.io/blog/streaming-in-the-clouds-where-to-start
• 

  Using Apache Kafka to Drive Cutting-Edge Machine Learning
  cloud ksql stream processing architecture machine learning ecosystem model deployment • • sorangutan  

  1
  1
  Posts
  122
  Views

  

  Machine learning and the Apache Kafka^® ecosystem are a great combination for training and deploying analytic models at scale. I had previously discussed potential use cases and architectures for machine learning in mission-critical, real-time applications that leverage the Apache Kafka ecosystem as a scalable and reliable central nervous system for your data. Let’s recap the architecture:We’ll expand on that point of view in this blog post by discussing new trends and innovations around Apache Kafka and machine learning, and how they are related. Let’s take a look!

  Overview

  • Embedded model deployment
    • Model deployment in Kafka applications
  • Hybrid cloud and on-prem architectures
    • Building hybrid cloud architectures with Kafka
  • Real-time streaming analytics
    • KSQL and machine learning for preprocessing and model deployment
  • Simplify building analytic models with AutoML
    • AutoML and the Apache Kafka ecosystem
  • Scalable and flexible machine learning platforms
    • Apache Kafka ecosystem for machine learning platforms
  • Apache Kafka as a key to flexible and future-ready machine learning

  Embedded model deployment

  On a high level, a machine learning lifecycle contains of two different parts:

  • Model training: In this step, we feed historical data into an algorithm to learn patterns from the past. The result is an analytic model.
  • Generating predictions: In this step, we use an analytic model for making predictions on new events based on the learned pattern.

  Machine learning is a continuous process, where we repeatedly improve and redeploy the analytic model over time.

  Predictions can be performed in different ways within an application or microservice. One way is to embed an analytic model directly into a stream processing application, like an application that uses Kafka Streams. You could, for example, use the TensorFlow for Java API to load and apply models:Alternatively, you could deploy the analytic models to a dedicated model server (like TensorFlow Serving), and use RPCs from the streaming application to the service (e.g., with HTTP or gRPC): Both options have various tradeoffs.

  Pros of a dedicated model server:

  • Simple integration with existing technologies and organizational processes
  • Easier to understand if you come from the non-streaming world
  • Later migration to real streaming is also possible
  • Model management built-in for deployment of different models including versioning, A/B testing, etc.

  Cons of a dedicated model server:

  • Often tied to specific machine learning technologies
  • Often tied to a specific cloud provider and introduces lock-in
  • Higher latency
  • More complex security setups (remote communication through firewalls and authorization management)
  • No offline inference (devices, edge processing, etc.)
  • Couples the availability, scalability and latency/throughput of your stream processing application with the SLAs of the RPC interface
  • Side effects (e.g., in case of failure) not covered by Kafka processing (e.g., exactly once)

  Whether or not to go with a model serving infrastructure is a question that often comes up for various reasons, including latency and security considerations. Some of the trends discussed below like hybrid architectures or low latency requirements also force you to think about model deployment, and whether you should deploy models locally at the edge to process personally identifiable information (PII), or if you want to integrate with external AutoML services.

  These two options and their tradeoffs are very important to understand so you can pick what is best for your use case and architecture.

  Model deployment in Kafka applications

  Kafka applications are event based, and leverage stream processing to continuously process input data. If you’re using Kafka, then you can embed an analytic model natively in a Kafka Streams or KSQL application. There are various examples of Kafka Streams microservices embedding models built with TensorFlow, H2O or Deeplearning4j natively.

  It is not always possible or feasible to embed analytic models directly due to architectural, security or organizational reasons. You can also choose to use RPC to perform model inference from your Kafka application (bearing in mind the the pros and cons discussed above). You can visit my project for an example of gRPC integration between a Kafka Streams microservice and locally hosted TensorFlow Serving container for making predictions with a hosted TensorFlow model.

  Hybrid cloud and on-prem architectures

  Hybrid cloud architectures are very common for machine learning infrastructure. Training can happen with large sets of historical data in the public cloud or in a central data lake within your own datacenter. Model inference can be done anywhere.

  In many scenarios, it totally makes sense to leverage the extreme scale and elasticity of public clouds. For example, you can spin up new large computing instances to train a neural network for a few days, then stop the instances. Pay-as-you-go is a perfect model, especially for deep learning.

  In the cloud you can leverage specific processing units which would be expensive and often idle in your own datacenter. For example, Google’s TPU (Tensor Processing Unit)—an application-specific integrated circuit (ASIC) designed from the ground up for machine learning—is one specific processor built just for deep learning. TPUs do one thing well: matrix multiplication—the heart of deep learning—to train neural networks. Compare this to CPUs or GPUs which are used in a broader landscape.

  If you need to keep data out of the public cloud, or if you want to build your own analytic infrastructure for bigger teams or departments, you might buy purpose-built hardware/software combinations for deep learning, such as Nvidia® DGX platforms for your own datacenter

  Wherever you need it, model inference can be performed independent of the model training: in the public cloud, on prem in your data centers or on edge devices like Internet of Things (IoT) or mobile devices. Edge devices often have higher latency, limited bandwidth or poor connectivity.

  Building hybrid cloud architectures with Kafka

  Apache Kafka enables you to build cloud-independent infrastructure—either multi-cloud or hybrid architectures. You are not bound to specific cloud APIs or proprietary products. Here is an example of a hybrid Kafka infrastructure for training and deploying analytic models:You can use Apache Kafka components like Kafka Connect for data ingestion and Kafka Streams or KSQL for the preprocessing of data. Model inference can also be done within a Kafka client including KSQL. The whole monitoring of the analytic infrastructure is often done with Apache Kafka, too. This includes technical metrics like latency and business information like model accuracy.

  Often the data streams come from other data centers or clouds. A common scenario is to use a Kafka replication tool like MirrorMaker or Confluent Replicator to replicate the data from source Kafka clusters to the analytics environment in a reliable and scalable way.

  Setup, configuration and operations of a reliable and scalable Kafka cluster requires a solid understanding of distributed systems and experience in running them.

  As an alternative you can use a cloud service like Confluent Cloud, which provides Kafka as a service. You then build the Kafka client (e.g., Kafka’s Java API, Kafka Streams or KSQL) and use the Kafka server side as an API. In your own data centers, tools like Confluent Operator and Confluent Control Center help to monitor, operate and manage your Kafka clusters.

  Unleashing Apache Kafka and TensorFlow in the Cloud is a helpful resource for learning more about hybrid machine learning infrastructures leveraging the Apache Kafka ecosystem, and Confluent Cloud on GCP and AWS.

  Real-time streaming analytics

  A primary requirement for many use cases is to process information whilst it is still relevant. It is important for all parts of a machine learning infrastructure:

  • (Re-)Training analytic models with current information
  • Predictions on new events in real time (often within milliseconds or seconds)
  • Monitoring the whole infrastructure for model accuracy, infrastructure errors and other details
  • Security-related tracking information like access control, auditing or provenance

  The easiest and most reliable way to process data in a timely manner requires building real-time streaming analytics natively on top of Apache Kafka using Kafka clients such as Java, .NET, Go or Python. As well as using a Kafka client, you should look at using Kafka Streams, the stream processing runtime for Apache Kafka. This is a Java library that enables you to perform simple (and complex) stream processing within your Java applications.

  If you don’t want to write Java or Scala, then there is also KSQL. KSQL is the streaming SQL engine for Apache Kafka. With KSQL you can write stream processing applications expressed in SQL. Read Neha Narkhede’s blog post Introducing KSQL: Open Source Streaming SQL for Apache Kafka for more details. To get an idea of things that you can do with KSQL, have a look at the KSQL recipes, and our KSQL blogs.

  KSQL and machine learning for preprocessing and model deployment

  Processing streaming data with KSQL makes data preparation for machine learning both easy and scalable. You can write SQL statements to do filtering, enrichments, transformations, feature engineering or other tasks. Here is just one example to filter sensor data from multiple types of cars for a specific car model for further processing or analytics:

  Machine learning models are easily embedded in KSQL by building a user-defined function (UDF). I created the example Deep Learning UDF for KSQL for Streaming Anomaly Detection of MQTT IoT Sensor Data, where I apply a neural network—more precisely an unsupervised autoencoder—for sensor analytics to detect anomalies:

  

  In this example, KSQL continuously processes millions of events from connected cars via MQTT integration to the Kafka cluster. MQTT is a publish / subscribe messaging protocol built for constrained devices and unreliable networks. It is often used in combination with Apache Kafka to integrate IoT devices with the rest of the enterprise. The autoencoder is used for predictive maintenance.

  Real-time analytics combined with this car sensor data allows you to send anomalies to a warning or emergency system to act before the engine breaks. Other use cases for intelligent connected cars worth noting include optimized routings and logistics, selling new features for a better digital car experience and loyalty services that integrate with restaurants and other shops on the streets.

  While a KSQL UDF requires writing a bit of code, this has to be done only once by the developer. Afterwards, the end user can simply use the UDF within their KSQL statements like any built-in function. Here is the KSQL query from this example, using the ANOMALY UDF, which applies the TensorFlow model under the hood:

  Both KSQL and Kafka Streams, depending on your preference and requirements, are a perfect fit for machine learning infrastructures when it comes to preprocessing streaming data and doing model inference. KSQL lowers the entry barrier and allows you to realize streaming applications with simple SQL statements instead of writing source code.

  Simplify building analytic models with AutoML

  AutoML, a recent development in machine learning, allows you to automatically build different machine learning algorithms with various hyperparameter settings and features to choose and deploy the best analytic model. With AutoML, you can build analytic models without any knowledge about machine learning itself.

  AutoML uses different implementations of algorithms like decision trees, clustering or neural networks to build and compare different models out of the box. You just upload or connect your historical datasets and click a few buttons to start the process.

  AutoML is not perfect for every use case today, but you can easily improve many existing business processes without needing an expensive data scientist, which many organizations cannot afford. Of course, some experience and knowledge in machine learning or data processing is helpful. Even more important, however, is domain knowledge to interpret the data and models correctly.

  DataRobot or Google’s Cloud AutoML^BETA are two of several public cloud offerings in this space. H2O’s AutoML is integrated into its open source machine learning framework, but they also offer a nice UI-focused commercial product called Driverless AI. I highly recommend spending at least 30 minutes to check out any AutoML tool, as it is really fascinating to see how AutoML tools evolve these days, even though AutoML is early on the maturity curve.

  AutoML and the Apache Kafka ecosystem

  One of the major advantages of using KSQL is that it enables you to process data and apply analytic models without writing complex source code in programming languages like Java. Instead, KSQL provides users with a SQL-like syntax in which to build stream processing applications. Together, KSQL and AutoML greatly simplify the process of building and deploying analytic models.

  Most AutoML tools include a model server for deployment of their models. One way that you can access the analytic models is via a REST interface. As discussed in an earlier section, although an RPC call for making predictions is not a perfect solution for a scalable, event-driven architecture like Kafka, there is an alternative approach: several AutoML solutions allow you to export their generated models so that you can deploy them into your application.

  AutoML in H2O’s open source framework can be applied via several interfaces like R, Python, Scala and a web UI. Here is an example in R:

  It’s similar to what you would do with H2O’s Python API to build a logistic regression, decision tree or neural network in H2O. But this time, you use the higher level automl function, which automates many of the typical tasks. The result is generated Java code that you can easily embed into your Kafka Streams microservice, KSQL application (via a UDF) or any other Kafka client, as can be seen through TensorFlow or Deeplearning4j examples.

  Scalable and flexible machine learning platforms

  Tech giants are typically some years ahead of “traditional enterprises.” They already built what you (have to) build today or tomorrow. Machine learning platforms are no exception.

  The paper Hidden Technical Debt in Machine Learning Systems explains why building and deploying an analytic model is much more effort than just writing some machine learning code with technologies like Python and TensorFlow. You also need to take care of data collection, feature extraction, serving infrastructure, monitoring and other tasks—all by using a scalable and reliable infrastructure:

  Figure 1. Hidden Technical Debt in Machine Learning Systems

  In addition, tech giants show that one machine learning/deep learning framework is not sufficient for their use cases, and that machine learning is a fast growing space. A flexible machine learning architecture needs to support different technologies and frameworks. It also needs to scale well and be reliable if used for the most important business processes.

  That’s why many tech giants have built their own machine learning platforms, like Uber’s Michelangelo, Netflix’s Meson and Paypal’s fraud detection platform. These machine learning platforms allow them to build and monitor powerful, scalable analytic models, but also stay flexible to choose the right machine learning technology for each use case.

  Apache Kafka ecosystem for machine learning platforms

  One of the reasons why Apache Kafka is so successful is due to its adoption by many tech giants.

  Almost all the great Silicon Valley companies like LinkedIn, Netflix, Uber, eBay—you name it—blog and speak about their usage of Kafka as an event-driven central nervous system for their mission-critical applications. Many focus on the distributed streaming platform for messaging, but we also see more and more adoption of components like Kafka Connect, Kafka Streams, REST Proxy, Schema Registry and KSQL.

  As discussed above in this post, Kafka is a logical fit for a machine learning platform. Training, monitoring, deployment, inference, configuration, A/B testing, etc. That’s probably why Uber, Netflix and many others use Kafka already as a central component in their machine learning infrastructure, too.

  Kafka allows easy, simple deployments of machine learning tasks without the need for another big data cluster, yet it is flexible in integrating with other systems. If you rely on your batch data processing in Hadoop or want to do machine learning processing with Spark, just connect it to Kafka via Kafka Connect.

  For some real-world machine learning architectures, take a look at projects built with the Confluent Platform:

  • Severstal, a manufacturing company, which uses Kafka and machine learning for predictive maintenance to reduce downtime with real-time streaming data
  • Celmatix, a biotech company, uses Kafka and machine learning to make informed and proactive reproductive health decisions by leveraging real-time genomics data
  • Centrica Hive, a company providing solutions and services for connected homes, leverages Kafka Streams and machine learning to create predictive heating failure alerts and to optimize schedule predictions

  With the Kafka ecosystem as base of your machine learning infrastructure, you are not forced to use just one specific technology. One of the great design concepts of Kafka is that you can re-process data again and again from its distributed commit log:
  You can use different technologies for training analytic models, deploy the models to any Kafka-native or external client applications, build a central monitoring and auditing system and stay ready for future innovations in the machine learning space. All at scale, reliable and fault tolerant, as decoupled systems, loosely connected with Apache Kafka as the nervous system.

  Apache Kafka as key to flexible and future-ready machine learning infrastructure

  The Apache Kafka ecosystem is a perfect match for a machine learning architecture. Pick the components you need to build a scalable, reliable platform that is independent of a specific on-prem/cloud infrastructure or machine learning technology.

  You can build an integration pipeline for model training and deploy the analytic models for real-time inference and monitoring within Kafka applications. This does not change with new trends like hybrid architectures or AutoML. Quite the opposite, in fact: With Kafka as the central but distributed and scalable layer, you stay flexible and ready for new technologies and concepts of the future.

  Please let me know what you think. Do you already use Kafka in the machine learning space? What components in addition to Kafka do you use? Share your thoughts in a comment below.

  And if you haven’t already, be sure to download Confluent Platform, the leading distribution of Apache Kafka, to get started with KSQL.

  
  
  https://www.confluent.io/blog/using-apache-kafka-drive-cutting-edge-machine-learning
• 

  Kafka Connect Deep Dive – Converters and Serialization Explained
  ksql troubleshooting kafka connect connecting to apache kafk json data serialization • • sorangutan  

  1
  1
  Posts
  116
  Views

  

  Kafka Connect is part of Apache Kafka^®, providing streaming integration between data stores and Kafka. For data engineers, it just requires JSON configuration files to use. There are connectors for common (and not-so-common) data stores out there already, including JDBC, Elasticsearch, IBM MQ, S3 and BigQuery, to name but a few.

  For developers, Kafka Connect has a rich API in which additional connectors can be developed if required. In addition to this, it also has a REST API for configuration and management of connectors.

  Kafka Connect is modular in nature, providing a very powerful way of handling integration requirements. Some key components include:

  • Connectors – the JAR files that define how to integrate with the data store itself
  • Converters – handling serialization and deserialization of data
  • Transforms – optional in-flight manipulation of messages

  One of the more frequent sources of mistakes and misunderstanding around Kafka Connect involves the serialization of data, which Kafka Connect handles using converters. Let’s take a good look at how these work, and illustrate some of the common issues encountered.

  Overview

  • Kafka messages are just bytes
    • Choosing a serialization format
    • If I write to my target in JSON, must I use JSON for my topics?
  • Configuring converters
  • JSON and schemas
  • Common errors
    • Problem: Reading non-JSON data with JsonConverter
    • Problem: Reading non-Avro data with AvroConverter
    • Problem: Reading a JSON message without the expected schema/payload structure
  • Troubleshooting tips
    • Finding the Connect worker log
    • Finding the Kafka Connect configuration file
    • Inspecting a Kafka topic
    • If you think you’ve got string/JSON data…
    • If you think you’ve got Avro data…
  • Internal converters
  • Applying a schema to messages without a schema
  • Conclusion
  • Interested in more?

  Kafka messages are just bytes

  Kafka messages are organized into topics. Each message is a key/value, but that is all that Kafka requires. Both key and value are just bytes when they are stored in Kafka. This makes Kafka applicable to a wide range of use cases, but it also means that developers have the responsibility of deciding how to serialize the data.

  In configuring Kafka Connect, one of the key things to standardize on is the serialization format. You need to make sure that anyone reading from the topic is using the same serialization format as those writing to the topic. Otherwise, confusion and errors will ensue!

  

  There are various serialization formats with common ones including:

  • JSON
  • Avro
  • Protobuf
  • String delimited (e.g., CSV)

  There are advantages and disadvantages to each of these—well, except delimited, in which case it’s only disadvantages

  Choosing a serialization format

  Some guiding principles for choosing a serialization format include:

  • Schema. A lot of the time your data will have a schema to it. You may not like the fact, but it’s your responsibility as a developer to preserve and propagate this schema. The schema provides the contract between your services. Some message formats (such as Avro and Protobuf) have strong schema support, whilst others have lesser support (JSON) or none at all (delimited string).
  • Ecosystem compatibility. Avro is a first-class citizen in the Confluent Platform, with native support from the  Confluent Schema Registry, Kafka Connect, KSQL and more. Protobuf, on the other hand, relies on community contributions for partial levels of feature support.
  • Message size. Whilst JSON is plain text and relies on any compression configured in Kafka itself, Avro and Protobuf are both binary formats and thus provide smaller message sizes.
  • Language support. Support for Avro is strong in the Java space, but if your company is not Java based then you may find it less easy to work with.

  If I write to my target in JSON, must I use JSON for my topics?

  No, not at all. The format in which you’re reading data from a source, or writing it to an external data store, doesn’t need to have a bearing on the format you use to serialize your messages in Kafka.

  A connector in Kafka Connect is responsible for taking the data from the source data store (for example, a database) and passing it as an internal representation of the data to the converter. Kafka Connect’s converters then serialize this source data object onto the topic.

  The same happens in reverse when using Kafka Connect as a sink—the converter deserializes the data from the topic into this internal representation, which is passed to the connector to write to the target data store using the appropriate method specific to the target.

  What this means is that you can have data on a topic in Avro (for example), and when you come to write it to HDFS (for example), you simply specify that you want the sink connector to use that format.

  Configuring converters

  Kafka Connect takes a default converter configuration at the worker level, and it can also be overridden per connector. Since using the same serialization format throughout your pipelines is generally a good idea, you’ll often just set the converter at the worker, and never need to specify it in a connector. But maybe you’re pulling data from someone else’s topic and they’ve decided to use a different serialization format—in that case you’d set this in the connector configuration. Even though you override it in the connector’s configuration, it’s still the converter that performs the work.

  Connectors that are written properly never [de]serialize the messages stored in Kafka, and always let the configured converter do that work.

  
  Remember, Kafka messages are just pairs of key/value bytes, and you need to specify the converter for both keys and value, using the key.converter and value.converter configuration setting. In some situations, you may use different converters for the key and the value.

  

  Here’s an example of using the String converter. Since it’s just a string, there’s no schema to the data, and thus it’s not so useful to use for the value:

  "key.converter": "org.apache.kafka.connect.storage.StringConverter",

  Some converters have additional configuration. For Avro, you need to specify the Schema Registry. For JSON, you need to specify if you want Kafka Connect to embed the schema in the JSON itself. When you specify converter-specific configurations, always use the key.converter. or value.converter. prefix. For example, to use Avro for the message payload, you’d specify the following:

  "value.converter": "io.confluent.connect.avro.AvroConverter",
  "value.converter.schema.registry.url": "http://schema-registry:8081",

  Common converters include:

  • Avro – open source from Confluent

    io.confluent.connect.avro.AvroConverter

  • String – part of Apache Kafka

    org.apache.kafka.connect.storage.StringConverter

  • JSON – part of Apache Kafka

    org.apache.kafka.connect.json.JsonConverter

  • ByteArray – part of Apache Kafka

    org.apache.kafka.connect.converters.ByteArrayConverter

  • Protobuf – open source from the community ( to our friends at Blue Apron)

    com.blueapron.connect.protobuf.ProtobufConverter

  JSON and schemas

  Whilst JSON does not by default support carrying a schema, Kafka Connect does support a particular format of JSON in which the schema is embedded. The resulting data size can get large as the schema is included in every single message along with the schema.

  If you’re setting up a Kafka Connect source and want Kafka Connect to include the schema in the message it writes to Kafka, you’d set:

  value.converter=org.apache.kafka.connect.json.JsonConverter
  value.converter.schemas.enable=true

  The resulting message to Kafka would look like the example below, with schema and payload top-level elements in the JSON:

  {
    "schema": {
      "type": "struct",
      "fields": [
        {
          "type": "int64",
          "optional": false,
          "field": "registertime"
        },
        {
          "type": "string",
          "optional": false,
          "field": "userid"
        },
        {
          "type": "string",
          "optional": false,
          "field": "regionid"
        },
        {
          "type": "string",
          "optional": false,
          "field": "gender"
        }
      ],
      "optional": false,
      "name": "ksql.users"
    },
    "payload": {
      "registertime": 1493819497170,
      "userid": "User_1",
      "regionid": "Region_5",
      "gender": "MALE"
    }
  }
  

  Note the size of the message, as well as the proportion of it that is made up of the payload vs. the schema. Considering that this is repeated in every message, you can see why a format like Avro makes a lot of sense, as the schema is stored separately and the message holds just the payload (and is compressed at that).

  If you’re consuming JSON data from a Kafka topic into a Kafka Connect sink, you need to understand if the data already includes a schema. If it does—and it’s in the same format as above, not some arbitrary schema-inclusion format—then you’d set:

  value.converter=org.apache.kafka.connect.json.JsonConverter
  value.converter.schemas.enable=true

  However, if you’re consuming JSON data and it doesn’t have the schema/payload construct, such as this sample:

  {
    "registertime": 1489869013625,
    "userid": "User_1",
    "regionid": "Region_2",
    "gender": "OTHER"
  }

  …you must tell Kafka Connect not to look for a schema by setting schemas.enable=false:

  value.converter=org.apache.kafka.connect.json.JsonConverter
  value.converter.schemas.enable=false

  As before, remember that the converter configuration option (here, schemas.enable) needs the prefix of key.converter or value.converter as appropriate.

  Common errors

  Here are some of the common errors you can get if you misconfigure the converters in Kafka Connect. These will show up in the sinks you configure for Kafka Connect, as it’s this point at which you’ll be trying to deserialize the messages already stored in Kafka. Converter problems tend not to occur in sources because it’s in the source that the serialization is set. Each of these will cause the connector to fail, with a headline error of:

  ERROR WorkerSinkTask{id=sink-file-users-json-noschema-01-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask)
  org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler
     at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator. execAndHandleError(RetryWithToleranceOperator.java:178)
     at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute (RetryWithToleranceOperator.java:104)
  

  After this error, you’ll see a further stack trace describing exactly why it erred. Be aware that the above error will be thrown for any fatal error in a connector, so you may well see this for errors unrelated to serialization. To quickly visualize what errors you can expect with which misconfiguration, here’s a quick reference:

  

  Problem: Reading non-JSON data with JsonConverter

  If you have non-JSON data on your source topic but try to read it with the JsonConverter, you can expect to see:

  org.apache.kafka.connect.errors.DataException: Converting byte[] to Kafka Connect data failed due to serialization error:
  …
  org.apache.kafka.common.errors.SerializationException: java.io.CharConversionException: Invalid UTF-32 character 0x1cfa7e2 (above 0x0010ffff) at char #1, byte #7)

  This could be caused by the source topic being serialized in Avro or another format.

  Solution: If the data is actually in Avro, then change your Kafka Connect sink connector to use:

  "value.converter": "io.confluent.connect.avro.AvroConverter",
  "value.converter.schema.registry.url": "http://schema-registry:8081",

  OR if the topic is populated by Kafka Connect, and you can and would rather do so, switch the upstream source to emit JSON data:

  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "value.converter.schemas.enable": "false",

  Problem: Reading non-Avro data with AvroConverter

  This may be the most common error that I see reported again and again on places like the Confluent Community mailing list and Slack group. It happens when you try to use the Avro converter to read data from a topic that is not Avro. This would include data written by an Avro serializer other than the Confluent Schema Registry’s Avro serializer, which has its own wire format.

  org.apache.kafka.connect.errors.DataException: my-topic-name
  at io.confluent.connect.avro.AvroConverter.toConnectData(AvroConverter.java:97)
  …
  org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id -1
  org.apache.kafka.common.errors.SerializationException: Unknown magic byte!

  The solution is to check the source topic’s serialization format, and either switch Kafka Connect’s sink connector to use the correct converter, or switch the upstream format to Avro (which is a good idea). If the upstream topic is populated by Kafka Connect, then you can configure the source connector’s converter as follows:

  "value.converter": "io.confluent.connect.avro.AvroConverter",
  "value.converter.schema.registry.url": "http://schema-registry:8081",

  Problem: Reading a JSON message without the expected schema/payload structure

  As described earlier, Kafka Connect supports a special structure of JSON messages containing both payload and schema. If you try to read JSON data that does not contain the data in this structure, you will get this error:

  org.apache.kafka.connect.errors.DataException: JsonConverter with schemas.enable requires "schema" and "payload" fields and may not contain additional fields. If you are trying to deserialize plain JSON data, set schemas.enable=false in your converter configuration.

  To be clear, the only JSON structure that is valid for schemas.enable=true has schema and payload fields as the top-level elements (shown above).

  As the message itself states, if you just have plain JSON data, you should change your connector’s configuration to:

  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "value.converter.schemas.enable": "false",

  If you want to include the schema in the data, you can either switch to using Avro (recommended), or you can configure Kafka Connect upstream to include the schema in the message:

  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "value.converter.schemas.enable": "true",

  Troubleshooting tips

  Finding the Connect worker log

  To find the error log from Kafka Connect, you need to locate the Kafka Connect worker’s output. The location of this depends on how you launched Kafka Connect. There are several ways to install Kafka Connect, including Docker, Confluent CLI, systemd and manually from a downloaded archive. You’ll find the worker log in:

  • Docker: docker logs container_name
  • Confluent CLI: confluent log connect
  • systemd: Log file is written to /var/log/confluent/kafka-connect
  • Other: By default, Kafka Connect sends its output to stdout so you’ll find it in the terminal session that launched Kafka Connect

  Finding the Kafka Connect configuration file

  To change configuration properties for Kafka Connect workers (which apply to all connectors run), set the configuration accordingly:

  • Docker – Set environment variables, for example, in Docker Compose:

    CONNECT_KEY_CONVERTER: io.confluent.connect.avro.AvroConverter
    CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL: 'http://schema-registry:8081'
    CONNECT_VALUE_CONVERTER: io.confluent.connect.avro.AvroConverter
    CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: 'http://schema-registry:8081'

  • Confluent CLI – Use the configuration file etc/schema-registry/connect-avro-distributed.properties
  • systemd (deb/rpm) – Use the configuration file /etc/kafka/connect-distributed.properties
  • Other – When you launch Kafka Connect, you specify the worker properties file, for example:

    $ cd confluent-5.0.0
    $ ./bin/connect-distributed ./etc/kafka/connect-distributed.properties

  Inspecting a Kafka topic

  Let’s say we’ve hit one of the errors shown above, and want to troubleshoot why our Kafka Connect sink connector can’t read from a topic.

  We need to check the data on the topic that is being read, and ensure it’s in the serialization format that we think it is. Also, bear in mind that all the messages need to be in this format, so don’t just assume that because you’re now sending messages in the correct format to the topic there won’t be a problem. Existing messages on the topic are also read by Kafka Connect and other consumers.

  Below, I describe troubleshooting this from the command line, but there are a few other tools to be aware of:

  • Confluent Control Center includes the feature to visually inspect topic contents, including automagic determination of the serialization format
  • KSQL’s PRINT command will print the contents of a topic to the console, including automagic determination of the serialization format
  • The Confluent CLI tool includes the consume command, which can be used to read both string and Avro data

  If you think you’ve got string/JSON data…

  You can use console tools including kafkacat and kafka-console-consumer. My personal preference is kafkacat:

  $ kafkacat -b localhost:9092 -t users-json-noschema -C -c1
  {
    "registertime":1493356576434,"userid":"User_8","regionid":"Region_2","gender":"MALE"}

  Using the excellent jq, you can also validate and format the JSON:

  $ kafkacat -b localhost:9092 -t users-json-noschema -C -c1|jq '.'
  {
    "registertime": 1493356576434,
    "userid": "User_8",
    "regionid": "Region_2",
    "gender": "MALE"
  }

  If you get something like this, with a bunch of “weird” characters, chances are you’re looking at binary data, as would be written by something like Avro or Protobuf:

  $ kafkacat -b localhost:9092 -t users-avro -C -c1
  ڝ ��VUser_9Region_MALE

  If you think you’ve got Avro data…

  You should use a console tool designed for reading and deserializing Avro data. Here, I’m using kafka-avro-console-consumer. Make sure you specify the correct Schema Registry URL at which the schema is held:

  $ kafka-avro-console-consumer --bootstrap-server localhost:9092 \
                                --property schema.registry.url=http://localhost:8081 \
                                --topic users-avro \
                                --from-beginning --max-messages 1
  {"registertime":1505213905022,"userid":"User_5","regionid":"Region_4","gender":"FEMALE"}

  As before, you can pipe the resulting output through jq if you want to format it:

  $ kafka-avro-console-consumer --bootstrap-server localhost:9092 \
                                --property schema.registry.url=http://localhost:8081 \
                                --topic users-avro \
                                --from-beginning --max-messages 1 | \
                                jq '.'
  {
    "registertime": 1505213905022,
    "userid": "User_5",
    "regionid": "Region_4",
    "gender": "FEMALE"
  }

  Internal converters

  When run in distributed mode, Kafka Connect uses Kafka itself to store metadata about its operations, including connector configuration, offsets and so on.

  These Kafka topics themselves can be configured to use different converters through the internal.key.converter/internal.value.converter settings. However, these settings are purely for internal use, and indeed have been deprecated as of Apache Kafka 2.0. You should not change these, and you will get warnings from Apache Kafka as of version 2.0 if you do try to configure them.

  Applying a schema to messages without a schema

  A lot of the time, Kafka Connect will bring data in from places where there is already a schema, and retaining that schema is just a matter of using a suitable serialization format such as Avro. All downstream users of that data then benefit from the schema being available to them, with the compatibility guarantees that something like Schema Registry provides. But what if there is no explicit schema?

  Maybe you’re reading data from a flat file using the FileSourceConnector (which is not recommended for production, but is often used for PoCs). Or, perhaps you’re pulling data from a REST endpoint using the REST connector. Since both of these, along with others, have no inherent schema, it’s up to you to declare it.

  Sometimes you’ll just want to pass the bytes that you’ve read from the source through and put them on a topic. But most of the time, you’ll want to do the Right Thing and apply a schema so that the data can be used. Doing it once as part of the ingestion, instead of pushing the problem onto each consumer (potentially multiple), is a much better pattern to follow.

  You can write your own Kafka Streams application to apply schema to data in a Kafka topic, but you can also use KSQL. This post shows how to do it against JSON data pulled from a REST endpoint. Let’s look here at a simple example of applying a schema to some CSV data. Obviously to be able to do this, we have to know the schema itself!

  Let’s say we’ve got a Kafka topic testdata-csv with some CSV data in it. It looks like this:

  $ kafkacat -b localhost:9092 -t testdata-csv -C
  1,Rick Astley,Never Gonna Give You Up
  2,Johnny Cash,Ring of Fire

  By eyeballing it, we can guess at there being three fields, maybe something like:

  • ID
  • Artist
  • Song

  If we leave the data in the topic like this, then any application wanting to use the data—whether it’s a Kafka Connect sink, bespoke Kafka application or whatever—will need to guess this schema each time. Or, just as bad, the developer for each consuming application will need to constantly go back to the team providing the data to check about the schema and any changes to it. Just as Kafka decouples systems, this kind of schema dependency forces a hard coupling between teams, and it’s not a good thing.

  So what we’ll do is simply apply a schema to the data using KSQL, and populate a new, derived topic in which the schema is present. From KSQL, you can inspect the topic data:

  ksql> PRINT 'testdata-csv' FROM BEGINNING;
  Format:STRING
  11/6/18 2:41:23 PM UTC , NULL , 1,Rick Astley,Never Gonna Give You Up
  11/6/18 2:41:23 PM UTC , NULL , 2,Johnny Cash,Ring of Fire

  The first two fields here (11/6/18 2:41:23 PM UTC and NULL) are the timestamp and key of the Kafka message, respectively. The remaining fields are from our CSV file. Let’s register this topic with KSQL and declare the schema:

  ksql> CREATE STREAM TESTDATA_CSV (ID INT, ARTIST VARCHAR, SONG VARCHAR) \
  WITH (KAFKA_TOPIC='testdata-csv', VALUE_FORMAT='DELIMITED');
  
  Message
  ----------------
  Stream created
  ----------------

  Observe that KSQL now has a schema for the stream of data:

  ksql> DESCRIBE TESTDATA_CSV;
  
  Name                 : TESTDATA_CSV
   Field   | Type
  -------------------------------------
   ROWTIME | BIGINT (system)
   ROWKEY  | VARCHAR(STRING) (system)
   ID      | INTEGER
   ARTIST  | VARCHAR(STRING)
   SONG    | VARCHAR(STRING)
  -------------------------------------
  For runtime statistics and query details run: DESCRIBE EXTENDED <Stream,Table>;

  Check that the data is as expected by querying the KSQL stream. Note that at this point we’re simply acting as a Kafka consumer against the existing Kafka topic—we’ve not changed or duplicated any data yet.

  ksql> SET 'auto.offset.reset' = 'earliest';
  Successfully changed local property 'auto.offset.reset' from 'null' to 'earliest'
  ksql> SELECT ID, ARTIST, SONG FROM TESTDATA_CSV;
  1 | Rick Astley | Never Gonna Give You Up
  2 | Johnny Cash | Ring of Fire

  Finally, create a new Kafka topic, populated by the reserialized data with schema. KSQL queries are continuous, so in addition to sending any existing data from the source topic to the target one, KSQL will send any future data to the topic too.

  ksql> CREATE STREAM TESTDATA WITH (VALUE_FORMAT='AVRO') AS SELECT * FROM TESTDATA_CSV;
  
  Message
  ----------------------------
  Stream created and running
  ----------------------------

  Verify the data using the Avro console consumer:

  $ kafka-avro-console-consumer --bootstrap-server localhost:9092 \
                                  --property schema.registry.url=http://localhost:8081 \
                                  --topic TESTDATA \
                                  --from-beginning | \
                                  jq '.'
  {
    "ID": {
      "int": 1
  },
    "ARTIST": {
      "string": "Rick Astley"
  },
    "SONG": {
      "string": "Never Gonna Give You Up"
    }
  }
  […]

  You can even check the Schema Registry for the registered schema:

  $ curl -s http://localhost:8081/subjects/TESTDATA-value/versions/latest|jq '.schema|fromjson'
  {
    "type": "record",
    "name": "KsqlDataSourceSchema",
    "namespace": "io.confluent.ksql.avro_schemas",
    "fields": [
      {
        "name": "ID",
        "type": [
          "null",
          "int"
        ],
        "default": null
      },
      {
        "name": "ARTIST",
        "type": [
          "null",
          "string"
        ],
        "default": null
      },
      {
        "name": "SONG",
        "type": [
          "null",
          "string"
        ],
        "default": null
      }
    ]
  }

  Any new messages written to the original topic (testdata-csv) are automagically processed by KSQL, and written to the new TESTDATA topic in Avro. Now any application or team wanting to use this data can simply work with the TESTDATA topic, and take advantage of the Avro-serialized data with declared schema. You can also use this technique to change the number of partitions in a topic, the partitioning key and replication factor.

  Conclusion

  Kafka Connect is a very simple yet powerful tool to use for integrating other systems with Kafka. One of the most common sources of misunderstanding is the converters that Kafka Connect offers. We’ve covered that Kafka messages are just key/value pairs, and it’s important to understand which serialization you should use and then standardize on that in your Kafka Connect connectors.

  Interested in more?

  • For more information about Kafka Connect, you can refer to the documentation, as well as find help on the Confluent Community mailing list and Slack group
  • If you’d like to look at the code samples used in this blog, they are available on GitHub
  • The quick KSQL snippet shown above barely scratches the surface of what’s possible with KSQL, so definitely check out the KSQL recipes and blogs for more

  
  
  https://www.confluent.io/blog/kafka-connect-deep-dive-converters-serialization-explained
• 

  Apache Kafka Supports 200K Partitions Per Cluster
  apache kafka zookeeper connecting to apache kafk controller kafka cluster partitions • • sorangutan  

  1
  1
  Posts
  101
  Views

  

  In Kafka, a topic can have multiple partitions to which records are distributed. Partitions are the unit of parallelism. In general, more partitions leads to higher throughput. However, there are some factors that one should consider when having more partitions in a Kafka cluster. I am happy to report that the recent Apache Kafka 1.1.0 release has significantly increased the number of partitions that a single Kafka cluster can support from the deployment and the availability perspective.

  To understand the improvement, it’s useful to first revisit some of the basics about partition leaders and the controller. First, each partition can have multiple replicas for higher availability and durability. One of the replicas is designated as the leader and all the client requests are served from this lead replica. Second, one of the brokers in the cluster acts as the controller that manages the whole cluster. If a broker fails, the controller is responsible for selecting new leaders for partitions on the failed broker.

  The Kafka broker does a controlled shutdown by default to minimize service disruption to clients. A controlled shutdown has the following steps. (1) A SIG_TERM signal is sent to the broker to be shut down. (2) The broker sends a request to the controller to indicate that it’s about to shut down. (3) The controller then changes the partition leaders on this broker to other brokers and persists that information in ZooKeeper. (4) The controller sends the new leader to other brokers. (5) The controller sends a successful reply to the shutting down broker, which finally terminates its process. At this point, there is no impact to the clients since their traffic has already been moved to other brokers. This process is depicted in Figure 1 below. Note that step (4) and (5) can happen in parallel.

  

   

  Figure 1. (1) shutdown initiated on broker 1; (2) broker 1 sends controlled shutdown request to controller on broker 0; (3) controller writes new leaders in ZooKeeper; (4) controller sends new leaders to broker 2; (5) controller sends successful reply to broker 1.

  Before Kafka 1.1.0, during the controlled shutdown, the controller moves the leaders one partition at a time. For each partition, the controller selects a new leader, writes it to ZooKeeper synchronously and communicates the new leader to other brokers through a remote request. This process has a couple of inefficiencies. First, the synchronous writes to ZooKeeper have higher latency, which slows down the controlled shutdown process. Second, communicating the new leader one partition at a time adds many small remote requests to every broker, which can cause the processing of the new leaders to be delayed.

  In Kafka 1.1.0, we made significant improvements in the controller to speed up the controlled shutdown. The first improvement is using the asynchronous API when writing to ZooKeeper. Instead of writing the leader for one partition, waiting for it to complete and then writing another one, the controller submits the leader for multiple partitions to ZooKeeper asynchronously and then waits for them to complete at the end. This allows for request pipelining between the Kafka broker and the ZooKeeper server and reduces the overall latency. The second improvement is that the communication of the new leaders is batched. Instead of one remote request per partition, the controller sends a single remote request with the leader from all affected partitions.

  We also made significant improvement in the controller failover time. If the controller goes down, the Kafka cluster automatically elects another broker as the new controller. Before being able to elect the partition leaders, the newly-elected controller has to first reload the state of all partitions in the cluster from ZooKeeper. If the controller has a hard failure, the window in which a partition is unavailable can be as long as the ZooKeeper session expiration time plus the controller state reloading time. So reducing the state reloading time improves the availability in this rare event. Prior to Kafka 1.1.0, the reloading uses the synchronous ZooKeeper API. In Kafka 1.1.0, this is changed to also use the asynchronous API for better latency.

  We executed tests to evaluate the performance improvement of the controlled shutdown time and the controller reloading time. For both tests, we set up a 5 node ZooKeeper ensemble on different server racks.

  In the first test, we set up a Kafka cluster with 5 brokers on different racks. In that cluster, we created 25,000 topics, each with a single partition and 2 replicas, for a total of 50,000 partitions. So, each broker has 10,000 partitions. We then measured the time to do a controlled shutdown of a broker. The results are shown in the table below.

  	kafka 1.0.0	kafka 1.1.0
  controlled shutdown time	6.5 minutes	3 seconds

  A big part of the improvement comes from fixing a logging overhead, which unnecessarily logs all partitions in the cluster every time the leader of a single partition changes. By just fixing the logging overhead, the controlled shutdown time was reduced from 6.5 minutes to 30 seconds. The asynchronous ZooKeeper API change reduced this time further to 3 seconds. These improvements significantly reduce the time to restart a Kafka cluster.

  In the second test, we set up another Kafka cluster with 5 brokers and created 2,000 topics, each with 50 partitions and 1 replica. This makes a total of 100,000 partitions in the whole cluster. We then measured the state reloading time of the controller and observed a 100% improvement (the reloading time dropped from 28 seconds in Kafka 1.0.0 to 14 seconds in Kafka 1.1.0).

  With those improvements, how many partitions can one expect to support in Kafka? The exact number depends on factors such as the tolerable unavailability window, ZooKeeper latency, broker storage type, etc. As a rule of thumb, we recommend each broker to have up to 4,000 partitions and each cluster to have up to 200,000 partitions. The main reason for the latter cluster-wide limit is to accommodate for the rare event of a hard failure of the controller as we explained earlier. Note that other considerations related to partitions still apply and one may need some additional configuration tuning with more partitions.

  The improvement that we made in 1.1.0 is just one step towards our ultimate goal of making Kafka infinitely scalable. We have also made improvements on latency with more partitions in 1.1.0 and will discuss that in a separate blog. In the near future, we plan to make further improvements to support millions of partitions in a Kafka cluster.

  The controller improvement work in Kafka 1.1.0 is a true community effort. Over a period of 9 months, people from 6 different organizations helped out and made this happen. Onur Karaman led the original design, did the core implementation and conducted the performance evaluation. Manikumar Reddy, Prasanna Gautam, Ismael Juma, Mickael Maison, Sandor Murakozi, Rajini Sivaram and Ted Yu each contributed some additional implementation or bug fixes. More details can be found in KAFKA-5642 and KAFKA-5027. A big thank you to all the contributors!

  This article was originally published by the Apache Kafka Committee on The Apache Software Foundation blog.

  
  
  https://www.confluent.io/blog/apache-kafka-supports-200k-partitions-per-cluster
• 

  How to Work with Apache Kafka in Your Spring Boot Application
  apache kafka how to message delivery spring boot • • sorangutan  

  1
  1
  Posts
  103
  Views

  

  Choosing the right messaging system during your architectural planning is always a challenge, yet one of the most important considerations to nail. As a developer, I write applications daily that need to serve lots of users and process huge amounts of data in real time.

  Usually, I use Java with the Spring Framework (Spring Boot, Spring Data, Spring Cloud, Spring Caching, etc.) for this. Spring Boot is a framework that allows me to go through my development process much faster and easier than before. It has come to play a crucial role in my organization. As the number of our users quickly grew, we realized our apparent need for something that could process as many as 1,000,000 events per second.

  When we found Apache Kafka^®, we saw that it met our needs and could handle millions of messages quickly. That’s why we decided to try it. And since that moment, Kafka has been a vital tool in my pocket. Why did I choose it, you ask?

  Apache Kafka is:

  • Scalable
  • Fault tolerant
  • A great publish-subscribe messaging system
  • Capable of higher throughput compared with most messaging systems
  • Highly durable
  • Highly reliable
  • High performant

  That’s why I decided to use it in my projects. Based on my experience, I provide here a step-by-step guide on how to include Apache Kafka in your Spring Boot application so that you can start leveraging its benefits too.

  Prerequisites

  • This article requires you to have Confluent Platform
  • Manual install using ZIP and TAR archives
    • Download
    • Unzip it
    • Follow the step-by-step instructions, and you’ll get Kafka up and running in your local environment

  I recommend using the Confluent CLI for your development to have Apache Kafka and other components of a streaming platform up and running.

  What you’ll get out of this guide

  After reading this guide, you will have a Spring Boot application with a Kafka producer to publish messages to your Kafka topic, as well as with a Kafka consumer to read those messages.

  And with that, let’s get started!

  Table of contents

  Step 1: Generate our project
  Step 2: Publish/read messages from the Kafka topic
  Step 3: Configure Kafka through application.yml configuration file
  Step 4: Create a producer
  Step 5: Create a consumer
  Step 6: Create a REST controller

  Step 1: Generate our project

  First, let’s go to Spring Initializr to generate our project. Our project will have Spring MVC/web support and Apache Kafka support.

  

  Once you have unzipped the project, you’ll have a very simple structure. I’ll show you how the project will look like at the end of this article so you can easily follow the same structure. I’m going to use Intellij IDEA, but you can use any Java IDE.

  Step 2: Publish/read messages from the Kafka topic

  Now, you can see what it looks like. Let’s move on to publishing/reading messages from the Kafka topic.

  Start by creating a simple Java class, which we will use for our example: package com.demo.models;

  public class User {
  
  private String name;
  private int age;
  
  public User(String name, int age) {
  this.name = name;
  this.age = age;
      }
  }

  

  Step 3: Configure Kafka through application.yml configuration file

  Next, we need to create the configuration file. We need to somehow configure our Kafka producer and consumer to be able to publish and read messages to and from the topic. Instead of creating a Java class, marking it with @Configuration annotation, we can use either application.properties file or application.yml. Spring Boot allows us to avoid all the boilerplate code we used to write in the past, and provide us with much more intelligent way of configuring our application, like this:

  server: port: 9000
  spring:
     kafka:
       consumer:
          bootstrap: localhost:9092
          group-id: group_id
          auto-offset-reset: earliest
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  producer:
          bootstrap: localhost:9092
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer

  If you want to get more about Spring Boot auto-configuration, you can read this short and useful article.

  Step 4: Create a producer

  Creating a producer will write our messages to the topic.

  @Service
  public class Producer {
  
      private static final Logger logger = LoggerFactory.getLogger(Producer.class);
      private static final String TOPIC = "users";
  
      @Autowired
      private KafkaTemplate<String, String> kafkaTemplate;
  
      public void sendMessage(String message) {
          logger.info(String.format("#### -> Producing message -> %s", message));
          this.kafkaTemplate.send(TOPIC, message);
      }
  }

  We just auto-wired KafkaTemplate and will use this instance to publish messages to the topic—that’s it for producer!

  Step 5: Create a consumer

  Consumer is the service that will be responsible for reading messages processing them according to the needs of your own business logic. To set it up, enter the following:

  @Service
  public class Consumer {
  
      private final Logger logger = LoggerFactory.getLogger(Producer.class);
  
      @KafkaListener(topics = "users", groupId = "group_id")
      public void consume(String message) throws IOException {
          logger.info(String.format("#### -> Consumed message -> %s", message));
      }
  }

  Here, we told our method void consume (String message) to subscribe to the user’s topic and just emit every message to the application log. In your real application, you can handle messages the way your business requires you to.

  Step 6: Create a REST controller

  If we already have a consumer, then we already have all we need to be able to consume Kafka messages.

  To fully show how everything that we created works, we need to create a controller with single endpoint. The message will be published to this endpoint, and then handled by our producer.

  Then, our consumer will catch and handle it the way we set it up  by logging to the console.

  @RestController
  @RequestMapping(value = "/kafka")
  public class KafkaController {
  
      private final Producer producer;
  
      @Autowired
      KafkaController(Producer producer) {
          this.producer = producer;
      }
  
      @PostMapping(value = "/publish")
      public void sendMessageToKafkaTopic(@RequestParam("message") String message) {
          this.producer.sendMessage(message);
      }
  }

  Let’s send our message to Kafka using cURL:

  curl -X POST -F 'message=test' http://localhost:9000/kafka/publish

  

  Basically, that’s it! In fewer than 10 steps, you learned how easy it is to add Apache Kafka to your Spring Boot project. If you followed this guide, you now know how to integrate Kafka into your Spring Boot project, and you are ready to go with this super tool!

  Interested in more?

  If you’d like to know more, you can download the Confluent Platform, the leading distribution of Apache Kafka. You can also find all the code in this article on GitHub.

  This is a guest post by Igor Kosandyak, a Java software engineer at Oril, with extensive experience in various development areas.

  
  
  https://www.confluent.io/blog/apache-kafka-spring-boot-application
• 

  ATM Fraud Detection with Apache Kafka and KSQL
  ksql stream processing apache kafka elastic anomaly detection elasticsearch • • sorangutan  

  1
  1
  Posts
  137
  Views

  

  Detecting fraudulent transactions is one of the classic use cases for real-time data. The business value is clear:

  • Reduce exposure to risk by identifying fraud sooner in order to take action to stop it
  • Improve customer experience by reducing false positives

  Traditionally, fraud detection has been done using stream processing applications written in languages such as Java. Maybe it’s been done using Kafka Streams or Spark Streaming, too, but regardless of the framework, it has always required a strong understanding of the core language first. This reduces the base of developers who can access stream processing. KSQL changes this, because anyone who can write SQL can now write stream processing applications.

  KSQL is a continuous query language. Whilst you can use it for interactive exploration of data—as we will see in a moment—its core purpose is building stream processing applications. Because KSQL is built on Kafka Streams, which is built on Apache Kafka^®, it benefits from the horizontal scalability characteristics for both resilience and throughput.

  We’re going to see how we can take a stream of inbound ATM transactions and easily set up an application to detect transactions that look fraudulent. We’ll use a three key factors to determine if we think a transaction is fraudulent:

  1. The same account number as a previous transaction…
  2. …but at a different location
  3. …less than 10 minutes after the previous one

   

  As part of the processing, we’ll also calculate the distance (as the crow flies) between the two suspect transactions, the time between them and, thus, how fast someone would have had to travel to plausibly make the transaction. This could feed into a second iteration of processing to refine the detection further.

  
  We’re going to use Elasticsearch to store the transactions and enriched data, and Kibana to provide visualization and further data inspection capabilities.

  Streams of ATM transactions

  KSQL has supported joining between two streams of events since Confluent Platform 5.0. This means you can take two Kafka topics and join between them on a common key. This could be something like a stream of order events and a stream of shipment events—with a stream-stream join, you can match up orders being placed to those being shipped.

  A stream-stream join can also make sense when done against the same stream of data, and that’s how we’re going to use it here. To start with, we’re going to take a simplified stream of ATM transactions and use it to understand what a stream-stream join is doing, and the semantics around it. Once we’ve done that, we’ll dive into a full-blown example with a data generator and some realistic test data.

  I’m using Docker Compose to easily provision a test environment—if you want to follow along, you can find all the example files on GitHub.

  Below are three transactions on three different accounts (ac_01, ac_02, ac_03). They have a transaction amount and location (both name and coordinates).

  {"account_id": "ac_01", "atm": "ID_2276369282", "amount": 20, "location": {"lat": "38.6956033", "lon": "-121.5922283"}, "transaction_id": "01"}
  {"account_id": "ac_02", "atm": "Flying Pig Bistro", "amount": 400, "location": {"lat": "37.766319", "lon": "-122.417422"}, "transaction_id": "02"}
  {"account_id": "ac_03", "atm": "Wells Fargo", "amount": 50, "location": {"lat": "37.5522855", "lon": "-121.9797997"}, "transaction_id": "04"}
  

  Let’s pull this into KSQL and see what we can do…

  ksql> SET 'auto.offset.reset'='earliest';
  ksql> CREATE STREAM ATM_TXNS (\
          account_id VARCHAR, \
          atm VARCHAR, \
          location STRUCT, \
          amount INT, \
          transaction_id VARCHAR) \
      WITH (\
          KAFKA_TOPIC='atm_txns', \
          VALUE_FORMAT='JSON');
  ksql> SELECT ACCOUNT_ID, TRANSACTION_ID, AMOUNT, ATM FROM ATM_TXNS;
  ac_01 | 01 | 20 | ID_2276369282
  ac_02 | 02 | 400 | Flying Pig Bistro
  ac_03 | 04 | 50 | Wells Fargo

  So far, so good. No naughty transactions—just three regular ones from three different accounts. Now, let’s fire in two more. The first is a second transaction at the same location for the same account (someone forgot to pick up the drinks bill!); the second one is our nefarious fraudster, with a clone of the card for account ac_03 drawing cash from an ATM elsewhere:

  {"account_id": "ac_02", "atm": "Flying Pig Bistro", "amount": 40, "location": {"lat": "37.766319", "lon": "-122.417422"}, "transaction_id": "03"}
  {"account_id": "ac_03", "atm": "Barclays", "amount": 500, "location": {"lat": "33.5522855", "lon": "-120.9797997"}, "transaction_id": "X05"}

  You’ll see these two new transactions appear straightaway in the KSQL output:

  ksql> SELECT ACCOUNT_ID, TRANSACTION_ID, AMOUNT, ATM FROM ATM_TXNS;
  ac_01 | 01 | 20 | ID_2276369282
  ac_02 | 02 | 400 | Flying Pig Bistro
  ac_03 | 04 | 50 | Wells Fargo
  ac_02 | 03 | 40 | Flying Pig Bistro
  ac_03 | X05 | 500 | Barclays
  

  ac_03 is the one we’re interested in, and the transaction ID has an X prefix to help us spot it in the output, and the key thing is that it’s the same account in a different location (Flying Pig Bistro and then Barclays). What about time window? I’m glad you asked!

  All Kafka messages have a timestamp (as well as a key and value). This can be set by the producing application, or it will otherwise take the value of the time at which the message arrived at the broker. KSQL exposes the message’s timestamp through the system column ROWTIME:

  ksql> SELECT TIMESTAMPTOSTRING(ROWTIME, 'yyyy-MM-dd HH:mm:ss'),ACCOUNT_ID, TRANSACTION_ID, AMOUNT, ATM FROM ATM_TXNS;
  2018-10-05 16:56:08 | ac_01 | 01 | 20 | ID_2276369282
  2018-10-05 16:56:08 | ac_02 | 02 | 400 | Flying Pig Bistro
  2018-10-05 16:56:08 | ac_03 | 04 | 50 | Wells Fargo
  2018-10-05 17:01:59 | ac_02 | 03 | 40 | Flying Pig Bistro
  2018-10-05 17:01:59 | ac_03 | X05 | 500 | Barclays
  

  The two transactions for account ac_03 occurred within the span of just over five minutes.
  By eyeballing this data, we can spot that something doesn’t look right; but how do we express this in stream processing terms? Enter stream-stream joins.

  Join the streams (but don’t cross them)

  Just as rows from two database tables can be joined, the same is possible with streams. The only difference is that since streams are unbounded (semantically—and often in practice—there is no defined end to the events arriving on the stream), the join syntax needs to account for this. Whilst a join between two tables in a database will consider all rows as candidates for matching on the join key, a stream-stream join will typically expect a window in which to look for events—which are, in effect, “rows”—to join. Otherwise, the stream processing engine would have to cache all events indefinitely.

  In KSQL, the time window is expressed using the WITHIN clause of the join. You can specify either a single window value which applies to either side of the event, or two window values, giving the range before and after the event in which the join is to be evaluated. This time window is a convenient vehicle for our fraud criteria—any two transactions within a 10-minute window are to be captured.

  We’ll begin by creating a second KSQL stream for the transaction events:

  CREATE STREAM ATM_TXNS_02 WITH (PARTITIONS=1) AS SELECT * FROM ATM_TXNS;

  Now we have two identical streams, one of which is actually a clone of the first:

  ksql> SHOW STREAMS;
  
   Stream Name | Kafka Topic | Format
  ------------------------------------
   ATM_TXNS    | atm_txns    | JSON
   ATM_TXNS_02 | ATM_TXNS_02 | JSON
  ------------------------------------
  
  ksql> SELECT TIMESTAMPTOSTRING(ROWTIME, 'yyyy-MM-dd HH:mm:ss'), ACCOUNT_ID,  TRANSACTION_ID, AMOUNT, ATM FROM ATM_TXNS_02;
  2018-10-05 16:56:08 | ac_03 | 04 | 50 | Wells Fargo
  2018-10-05 16:56:08 | ac_01 | 01 | 20 | ID_2276369282
  2018-10-05 17:01:59 | ac_03 | X05 | 500 | Barclays
  2018-10-05 16:56:08 | ac_02 | 02 | 400 | Flying Pig Bistro
  2018-10-05 17:01:59 | ac_02 | 03 | 40 | Flying Pig Bistro
  

  Note: This is a necessary step that in principle could be handled by KSQL. It is tracked in GitHub issue 2030.

  ksql> SELECT TIMESTAMPTOSTRING(T1.ROWTIME, 'yyyy-MM-dd HH:mm:ss'), TIMESTAMPTOSTRING(T2.ROWTIME, 'yyyy-MM-dd HH:mm:ss'), \
               T1.ACCOUNT_ID, T2.ACCOUNT_ID, \
               T1.TRANSACTION_ID, T2.TRANSACTION_ID, \
               T1.LOCATION, T2.LOCATION \
        FROM   ATM_TXNS T1 \
        INNER JOIN ATM_TXNS_02 T2 \
                   WITHIN 10 MINUTES \
                   ON T1.ACCOUNT_ID = T2.ACCOUNT_ID ;
  
  2018-10-05 16:56:08 | 2018-10-05 16:56:08 | ac_01 | ac_01 | 01 | 01 | {LON=-121.5922283, LAT=38.6956033} | {LON=-121.5922283, LAT=38.6956033}
  2018-10-05 16:56:08 | 2018-10-05 16:56:08 | ac_02 | ac_02 | 02 | 02 | {LON=-122.417422, LAT=37.766319} | {LON=-122.417422, LAT=37.766319}
  2018-10-05 16:56:08 | 2018-10-05 17:01:59 | ac_02 | ac_02 | 02 | 03 | {LON=-122.417422, LAT=37.766319} | {LON=-122.417422, LAT=37.766319}
  2018-10-05 16:56:08 | 2018-10-05 16:56:08 | ac_03 | ac_03 | 04 | 04 | {LON=-121.9797997, LAT=37.5522855} | {LON=-121.9797997, LAT=37.5522855}
  2018-10-05 16:56:08 | 2018-10-05 17:01:59 | ac_03 | ac_03 | 04 | X05 | {LON=-121.9797997, LAT=37.5522855} | {LON=-120.9797997, LAT=33.5522855}
  2018-10-05 17:01:59 | 2018-10-05 16:56:08 | ac_02 | ac_02 | 03 | 02 | {LON=-122.417422, LAT=37.766319} | {LON=-122.417422, LAT=37.766319}
  2018-10-05 17:01:59 | 2018-10-05 17:01:59 | ac_02 | ac_02 | 03 | 03 | {LON=-122.417422, LAT=37.766319} | {LON=-122.417422, LAT=37.766319}
  2018-10-05 17:01:59 | 2018-10-05 16:56:08 | ac_03 | ac_03 | X05 | 04 | {LON=-120.9797997, LAT=33.5522855} | {LON=-121.9797997, LAT=37.5522855}
  2018-10-05 17:01:59 | 2018-10-05 17:01:59 | ac_03 | ac_03 | X05 | X05 | {LON=-120.9797997, LAT=33.5522855} | {LON=-120.9797997, LAT=33.5522855}

  Looking at the output, we see a lot more than just the fraudulent transaction we’re expecting to identify. We can explain these additional matches as thus:

  Commentary	T1 timestamp	T2 timestamp	T1 account	T2 account	T1 TXN ID	T2 TXN ID	T1 Location	T2 Location
  self-join	2018-10-05 16:56:08	2018-10-05 16:56:08	ac_01	ac_01	01	01	{LON=-121.5922283, LAT=38.6956033}	{LON=-121.5922283, LAT=38.6956033}
  self-join	2018-10-05 16:56:08	2018-10-05 16:56:08	ac_02	ac_02	02	02	{LON=-122.417422, LAT=37.766319}	{LON=-122.417422, LAT=37.766319}
  self-join	2018-10-05 16:56:08	2018-10-05 17:01:59	ac_02	ac_02	02	03	{LON=-122.417422, LAT=37.766319}	{LON=-122.417422, LAT=37.766319}
  self-join	2018-10-05 16:56:08	2018-10-05 16:56:08	ac_03	ac_03	04	04	{LON=-121.9797997, LAT=37.5522855}	{LON=-121.9797997, LAT=37.5522855}
  !FRAUD!	2018-10-05 16:56:08	2018-10-05 17:01:59	ac_03	ac_03	04	X05	{LON=-121.9797997, LAT=37.5522855}	{LON=-120.9797997, LAT=33.5522855}
  valid (same location, not shown)	2018-10-05 17:01:59	2018-10-05 16:56:08	ac_02	ac_02	03	02	{LON=-122.417422, LAT=37.766319}	{LON=-122.417422, LAT=37.766319}
  self-join	2018-10-05 17:01:59	2018-10-05 17:01:59	ac_02	ac_02	03	03	{LON=-122.417422, LAT=37.766319}	{LON=-122.417422, LAT=37.766319}
  !FRAUD! (duplicate)	2018-10-05 17:01:59	2018-10-05 16:56:08	ac_03	ac_03	X05	04	{LON=-120.9797997, LAT=33.5522855}	{LON=-121.9797997, LAT=37.5522855}
  self-join	2018-10-05 17:01:59	2018-10-05 17:01:59	ac_03	ac_03	X05	X05	{LON=-120.9797997, LAT=33.5522855}	{LON=-120.9797997, LAT=33.5522855}

  
  The first thing to do is weed out the join results where it’s just the same event joining to itself (that is, the transaction ID is the same):

  ksql> SELECT
        […]
        WHERE T1.TRANSACTION_ID != T2.TRANSACTION_ID ;
  
  2018-10-05 17:01:59 | 2018-10-05 16:56:08 | ac_02 | ac_02 | 03 | 02 | {LON=-122.417422, LAT=37.766319} | {LON=-122.417422, LAT=37.766319}
  2018-10-05 17:01:59 | 2018-10-05 16:56:08 | ac_03 | ac_03 | X05 | 04 | {LON=-120.9797997, LAT=33.5522855} | {LON=-121.9797997, LAT=37.5522855}
  2018-10-05 16:56:08 | 2018-10-05 17:01:59 | ac_02 | ac_02 | 02 | 03 | {LON=-122.417422, LAT=37.766319} | {LON=-122.417422, LAT=37.766319}
  2018-10-05 16:56:08 | 2018-10-05 17:01:59 | ac_03 | ac_03 | 04 | X05 | {LON=-121.9797997, LAT=37.5522855} | {LON=-120.9797997, LAT=33.5522855}

  Much better. Now, we just need to eliminate the transactions on the same account that took place at the same location—in this model, our fraud criteria determine those as unsuspicious.

  ksql> SELECT
        […]
        WHERE
        […]
        (T1.location->lat != T2.location->lat OR \
         T1.location->lon != T2.location->lon);
  
  2018-10-05 16:56:08 | 2018-10-05 17:01:59 | ac_03 | ac_03 | 04 | X05 | {LON=-121.9797997, LAT=37.5522855} | {LON=-120.9797997, LAT=33.5522855}
  2018-10-05 17:01:59 | 2018-10-05 16:56:08 | ac_03 | ac_03 | X05 | 04 | {LON=-120.9797997, LAT=33.5522855} | {LON=-121.9797997, LAT=37.5522855}
  

  The only two results are those on the account ac_03, one being genuine (transaction ID 04) and the other fraudulent (X05). We’re getting both returned, as each is an event on the left hand stream (the driving one) that joins to the other based on the time window specified (10 minutes before or after the driving event). All we need to do is change our join window to only return events that happen after the one we’re using to drive the join. To do this, simply specify a zero BEFORE threshold in the WITHIN criteria:

  ksql> SELECT
        […]
        WITHIN (0 MINUTES, 10 MINUTES) \
        […]
  
  2018-10-05 16:56:08 | 2018-10-05 17:01:59 | ac_03 | ac_03 | 04 | X05 | {LON=-121.9797997, LAT=37.5522855} | {LON=-120.9797997, LAT=33.5522855}

  With the core logic of the statement built, let’s add in a few more bells and whistles. Using the built-in GEO_DISTANCE function, we can include a column in the output showing the distance between the two transactions:

  ksql> SELECT
        […]
        T1.location, \
        T2.location, \
        GEO_DISTANCE(T1.location->lat, T1.location->lon, T2.location->lat, T2.location->lon, 'KM') AS DISTANCE_BETWEEN_TXN_KM, \
        […]
  
  {LON=-121.9797997, LAT=37.5522855} | {LON=-120.9797997, LAT=33.5522855} | 453.87740037465375
  

  We see that transaction 04 took place over 450 kilometers, as the crow flies from X05. What was the time duration between them? We can answer this pretty easily based on the timestamp, but it’s more sensible just to include it in the query:

  ksql> SELECT
          […]
          (T2.ROWTIME - T1.ROWTIME) AS MILLISECONDS_DIFFERENCE,  \
          (CAST(T2.ROWTIME AS DOUBLE) - CAST(T1.ROWTIME AS DOUBLE)) / 1000 / 60 AS MINUTES_DIFFERENCE,  \
          (CAST(T2.ROWTIME AS DOUBLE) - CAST(T1.ROWTIME AS DOUBLE)) / 1000 / 60 / 60 AS HOURS_DIFFERENCE,  \
          GEO_DISTANCE(T1.location->lat, T1.location->lon, T2.location->lat, T2.location->lon, 'KM') / ((CAST(T2.ROWTIME AS DOUBLE) - CAST(T1.ROWTIME AS DOUBLE)) / 1000 / 60 / 60) AS KMH_REQUIRED,
          […]
  
  
  351473 | 5.8578833333333336 | 0.09763138888888889 | 4648.888083433872
  

  We’ve also combined the distance and time calculations to derive the speed at which someone would have to move between the two events. At 4,648 kph, it’s almost four times the speed of the fastest supersonic car—we can be pretty sure it’s fraudulent!

  One remaining point to make about the above query is that the message’s timestamp (ROWTIME) is cast from its BIGINT data type to DOUBLE so that the subsequent division arithmetic will work.

  Running it with “real” data

  To see what our query looks like against a continuous stream of transactions, let’s fire up our data generator. I’m using an open source tool called gess, which I’ve forked and tweaked to suit this demo.

  It works by taking a list of ATMs, generating transactions against them and emitting these to UDP. UDP is a networking protocol in the same way that TCP is, but unlike TCP, UDP doesn’t require any kind of acknowledgement of delivery. Instead, it just fires bytes out into the ether, and if someone picks them up, that’s great, and if not, that’s fine too. It makes for a useful test rig where we can start up the data generator and simply tap into the event stream as and when we want to.

  To route the events to Kafka from UDP, I’m using two great little command line tools that any self-respecting engineer should know: netcat (nc) and kafkacat. Netcat listens for the UDP traffic, which is then piped to kafkacat. kafkacat simply takes any input from stdin and sends it as messages to the target topic.

  Here’s netcat picking up the events:

  $ nc -v -u -l 6900
  {"account_id": "a9", "timestamp": "2018-10-07T20:40:48.585666", "atm": "ATM : 3616415159", "amount": 50, "location": {"lat": "53.8233994", "lon": "-1.4865327"}, "transaction_id": "e406bf57-ca68-11e8-a4cb-186590d22a35"}
  {"account_id": "a102", "timestamp": "2018-10-07T20:40:49.087221", "atm": "Co-op Bank", "amount": 400, "location": {"lat": "53.7986913", "lon": "-1.2518281"}, "transaction_id": "e4534754-ca68-11e8-a119-186590d22a35"}
  {"account_id": "a496", "timestamp": "2018-10-07T20:40:49.589651", "atm": "Link", "amount": 50, "location": {"lat": "53.8442149", "lon": "-1.5094248"}, "transaction_id": "e49ff142-ca68-11e8-9c4f-186590d22a35"}
  {"account_id": "a223", "timestamp": "2018-10-07T20:40:50.093244", "atm": "ATM : 5523013160", "amount": 400, "location": {"lat": "53.6781485", "lon": "-1.4991026"}, "transaction_id": "e4ecc8fd-ca68-11e8-9132-186590d22a35"}
  [...]

  Here’s netcat piping it to a Kafka topic:

  $ nc -v -u -l 6900 | \
      docker run --interactive --rm --network ksql-atm-fraud-detection_default confluentinc/cp-kafkacat \
      kafkacat -b kafka:29092 -P -t atm_txns_gess
  

  Note that there’s no console output from this, because it’s being redirected to kafkacat.

  Event time processing with KSQL

  We need to make one change to the KSQL statement that we developed above. Whereas we were previously using the Kafka message timestamp as the event rowtime, now, we want to use the timestamp field that’s included in the payload of the message. This is easy to do with KSQL by simply specifying the TIMESTAMP field in the WITH clause:

  CREATE STREAM ATM_TXNS_GESS (\
                account_id     VARCHAR, \
                atm            VARCHAR, \
                location       STRUCT<lon DOUBLE,  \
                                      lat DOUBLE>, \
                amount         INT, \
                timestamp      VARCHAR, \
                transaction_id VARCHAR) \
           WITH (\
                KAFKA_TOPIC='atm_txns_gess', \
                VALUE_FORMAT='JSON', \
                TIMESTAMP='timestamp', \
                TIMESTAMP_FORMAT='yyyy-MM-dd HH:mm:ss X');

  Just to check that KSQL is indeed picking up the value of timestamp field in the source message, let’s run a query to report the timestamp field’s value along with the system column ROWTIME, which represents the timestamp with which KSQL will process the message:

  ksql> SELECT TIMESTAMPTOSTRING(ROWTIME, 'yyyy-MM-dd HH:mm:ss Z'), timestamp FROM ATM_TXNS_GESS;
  2018-10-07 22:31:39 +0000 | 2018-10-07 22:31:39 +0000
  2018-10-07 22:31:40 +0000 | 2018-10-07 22:31:40 +0000
  2018-10-07 22:26:58 +0000 | 2018-10-07 22:26:58 +0000
  2018-10-07 22:31:41 +0000 | 2018-10-07 22:31:41 +0000

  As expected, they match. One subtlety to notice here is that the third message above is dated earlier than the one previously. That’s because the ATM transactions may be arriving out of order. However, KSQL will process them based on event time (i.e., timestamp value in the source message, when the actual ATM transaction occurred) rather than processing time, which refers to when the message arrived at the system.

  Bringing together our new source stream (ATM_TXNS_GESS) with the logic we prototyped above gives us this code to run:

  CREATE STREAM ATM_TXNS_GESS_02 WITH (PARTITIONS=1) AS SELECT * FROM ATM_TXNS_GESS;
  
  CREATE STREAM ATM_POSSIBLE_FRAUD  \
      WITH (PARTITIONS=1) AS \
  SELECT T1.ROWTIME AS T1_TIMESTAMP, T2.ROWTIME AS T2_TIMESTAMP, \
          GEO_DISTANCE(T1.location->lat, T1.location->lon, T2.location->lat, T2.location->lon, 'KM') AS DISTANCE_BETWEEN_TXN_KM, \
          (T2.ROWTIME - T1.ROWTIME) AS MILLISECONDS_DIFFERENCE,  \
          (CAST(T2.ROWTIME AS DOUBLE) - CAST(T1.ROWTIME AS DOUBLE)) / 1000 / 60 AS MINUTES_DIFFERENCE,  \
          GEO_DISTANCE(T1.location->lat, T1.location->lon, T2.location->lat, T2.location->lon, 'KM') / ((CAST(T2.ROWTIME AS DOUBLE) - CAST(T1.ROWTIME AS DOUBLE)) / 1000 / 60 / 60) AS KMH_REQUIRED, \
          T1.ACCOUNT_ID AS ACCOUNT_ID, \
          T1.TRANSACTION_ID, T2.TRANSACTION_ID, \
          T1.AMOUNT, T2.AMOUNT, \
          T1.ATM, T2.ATM, \
          CAST(T1.location->lat AS STRING) + ',' + CAST(T1.location->lon AS STRING) AS T1_LOCATION, \
          CAST(T2.location->lat AS STRING) + ',' + CAST(T2.location->lon AS STRING) AS T2_LOCATION \
  FROM   ATM_TXNS_GESS T1 \
         INNER JOIN ATM_TXNS_GESS_02 T2 \
          WITHIN (0 MINUTES, 10 MINUTES) \
          ON T1.ACCOUNT_ID = T2.ACCOUNT_ID \
  WHERE   T1.TRANSACTION_ID != T2.TRANSACTION_ID \
    AND   (T1.location->lat != T2.location->lat OR \
           T1.location->lon != T2.location->lon) \
    AND   T2.ROWTIME != T1.ROWTIME;
  
  

  Checking the output shows that there are plenty of fraudulent transactions being detected:

  SELECT T1_ACCOUNT_ID, \
         TIMESTAMPTOSTRING(T1_TIMESTAMP, 'yyyy-MM-dd HH:mm:ss'), TIMESTAMPTOSTRING(T2_TIMESTAMP, 'HH:mm:ss'), \
         T1_ATM, T2_ATM, \
         DISTANCE_BETWEEN_TXN_KM, MINUTES_DIFFERENCE \
  FROM ATM_POSSIBLE_FRAUD;
  
  a739 | 2018-10-08 15:35:58 | 15:38:31 | Halifax | Barclays Bank PLC | 15.698597512981406 | 2.55
  a649 | 2018-10-08 15:36:22 | 15:38:04 | Yorkshire Bank | Barclays Bank PLC | 23.179463348879413 | 1.7
  [...]
  

  
  The execution statistics shows that we’ve processed multiple messages. In other words, we’ve detected many possibly fraudulent transactions:

  ksql> DESCRIBE EXTENDED ATM_POSSIBLE_FRAUD;
  
  [...]
  Kafka topic : ATM_POSSIBLE_FRAUD (partitions: 1, replication: 1)
  [...]
  Local runtime statistics
  ------------------------
  messages-per-sec: 0.85 total-messages: 324 last-message: 10/9/18 11:25:11 AM UTC
  [...]
  

  There are some changes to note from the query that we iteratively built up at the beginning of this article. These are just to streamline and tidy it up, but the core logic is the same.

  • Add CREATE STREAM … AS to tell KSQL to persist this as a streaming application, and populate the named stream as a Kafka topic with the results.
  • Retain the timestamp as an epoch rather than the VARCHAR that we’ve been using for printing it in human-readable format.
  • Only include one of the ACCOUNT_ID fields in the output since they are equal, as stated in the JOIN criteria.
  • Remove the intermediate calculated columns of the time difference between the two transactions.
  • Include the name of the ATM at which each transaction took place.
  • Wrangle the source LOCATION column (a STRUCT by default) into a comma-separated STRING. This is necessary for being able to index it into Elasticsearch as a geopoint.

  Kafka + Elastic =

  Using the above KSQL application, we’ve got a Kafka topic being populated with suspect ATM transactions. We can query this from the command line in KSQL to inspect it, but at the end of the day, it’s just a Kafka topic. We can use this Kafka topic for multiple independent purposes, including:

  • Drive a microservice—perhaps to trigger an alert or block on a particular card
  • Stream the data to a store, such as Elasticsearch, for visualization and analysis in Kibana

  Streaming data from Kafka to Elasticsearch is easy using Kafka Connect and the Elasticsearch connector. Check out the code on GitHub for full details, but in essence, it’s two scripts:

  1. A dynamic mapping template for the Elasticsearch indices so that things like geopoints and timestamps are set up correctly
  2. A Kafka Connect JSON configuration specifying the Kafka topics from which to stream data and the corresponding Elasticsearch indices to load

  With the data in Elasticsearch, we can easily build some powerful dashboards and analyses with Kibana. Here’s a view of all suspected fraudulent activity in a region, with hotspots highlighted:

  
  By selecting a specific account, all ATM transactions for that account can be shown for further analysis. Here, any fraud alerts for account a410 are shown and plotted on the map:

  

  You can also use Kibana to draw a bounding box around a particular region of the map to filter events just for that area:
  

  Enriching streams of events

  We’ve taken an inbound stream of events and used KSQL to populate a Kafka topic of transactions that look potentially fraudulent. But all we have to go on is the account number. Wouldn’t it be more useful if we can include in our event stream information about the account itself? We can then show the account information in the visual analysis, as well as using it to drive notifications directly.

  Let’s assume that we have all of our customer information in a database. Pretty standard place to keep it. It’s maintained by a separate application, and it is our master store of customer data. In this case I’m using MySQL, but it could be any RDBMS, really:

  mysql> SELECT FIRST_NAME, LAST_NAME, EMAIL, ADDRESS FROM accounts WHERE ACCOUNT_ID='a42';
  +------------+-----------+--------------------+------------------+
  | FIRST_NAME | LAST_NAME | EMAIL              | ADDRESS          |
  +------------+-----------+--------------------+------------------+
  | Robin      | Moffatt   | [email protected] | 22 Acacia Avenue |
  +------------+-----------+--------------------+------------------+
  1 row in set (0.00 sec)
  

  Using Kafka Connect and a CDC tool such as Debezium, we can stream the contents of it to a Kafka topic, as well as any changes made to the data, in real time. With the data in a Kafka topic, it’s possible to model it as a table and join it to the event stream of ATM transactions:

  ksql> SELECT FIRST_NAME, LAST_NAME, EMAIL, ADDRESS FROM accounts WHERE ACCOUNT_ID='a42';
  Robin | Moffatt | [email protected] | 22 Acacia Avenue

  If something changes in the database, it’s reflected straight away in the Kafka topic (and thus KSQL table too):
  

  So with an accurate mirror of the data from the table available in KSQL, it’s a simple matter to join this to the stream of ATM transactions:

  ksql> SELECT A.ACCOUNT_ID, \
          C.FIRST_NAME + ' ' + C.LAST_NAME, \
          C.EMAIL, C.PHONE, \
          C.ADDRESS, \
          TIMESTAMPTOSTRING(A.T1_TIMESTAMP, 'yyyy-MM-dd HH:mm:ss'), TIMESTAMPTOSTRING(A.T2_TIMESTAMP, 'HH:mm:ss'), \
          A.T1_ATM, A.T2_ATM, \
          A.DISTANCE_BETWEEN_TXN_KM, A.MINUTES_DIFFERENCE \
  FROM ATM_POSSIBLE_FRAUD A LEFT OUTER JOIN ACCOUNTS C ON A.ACCOUNT_ID = C.ACCOUNT_ID;  
  a279 | Shandeigh Isakovic | [email protected] | +44 645 302 9358 | 49 Nevada Center | 2018-10-09 13:42:07 | 13:47:21 | Yorkshire Bank | ATM : 319429912 | 12.536338916950928 | 5.233333333333333
  a769 | Kathe Cutteridge | [email protected] | +44 501 421 3436 | 5 Jackson Pass | 2018-10-09 13:45:47 | 13:47:35 | Barclays | Co-Op | 14.448491852409132 | 1.8
  […]

  Persisting this as another new Kafka topic gives us a rich stream of events every time an ATM transaction occurs matching our fraud criteria. It also includes all the information we might want in order to contact the customer concerned:

  
  The same data is used for pushing fraud notifications to customers’ phones:

  

  Conclusion

  We’ve seen a few key concepts with KSQL:

  • Detecting patterns in a stream of data based on an event’s time relative to another’s, as well as characteristics of the two events
  • Populating a target Kafka topic with a real-time feed of these events matching the defined conditions
  • Denormalizing data, bringing together events (facts) and additional information about entities within those (dimensions)

  The enriched data from the Kafka topic, which is continually populated by KSQL, is streamed into Elasticsearch using Kafka Connect. From there, analysis and visualization are done in Kibana. This enriched topic serves as the driver for a microservice responsible for further actions based on suspected fraud on an account, such as putting a temporary hold on it or notifying the account holder.

  If you’re interested in what KSQL can do

  You can download the Confluent Platform, the leading distribution of Apache Kafka, to get started with KSQL.

  
  
  https://www.confluent.io/blog/atm-fraud-detection-apache-kafka-ksql
• 

  That’s a Wrap!Kafka Summit San Francisco 2018 Roundup
  kafka summit conferences • • sorangutan  

  1
  1
  Posts
  86
  Views

  

  Were you there last week? For as big as that event felt, it’s hard to believe it’s only the third annual Kafka Summit San Francisco. But the view was beautiful and the content was rich.

  

  Of course, given the size of the Apache Kafka^® Community, I can say for sure that most of you were not there. But that’s okay! Session videos are now online.

  And it’s always fun to look at the event by the numbers. Here’s what we had this time:

  • 4 keynotes
  • 20 different food trucks
  • 56 sessions over four tracks
  • 76 speakers
  • 350+ companies
  • 32 countries represented
  • 1,500 tweets made by 608 people
  • 2,816 images in the #kafkasummit mosaic
  • 8,000+ online views of the keynotes already

  Although 8,000 views of the keynotes is a lot, I know that still leaves some of you who haven’t watched them. I really do think they’re all worth your time:

  • In Overcome By Events, Confluent Co-founder and CEO Jay Kreps made the case the separate architectures usually employed to address microservices, monitoring, data pipeline and analytic applications really all point to the need for a streaming platform.
  • In What’s In Your Topic?, Capital One VP of Streaming Data walked us through how Capital One is building out a streaming platform at scale. The response to this keynote has been fantastic.
  • In Is Kafka a Database?, University of Cambridge Martin Kleppmann argues that Kafka has a stronger claim to ACID transactional semantics than any relational database does at scale. Provocative, but compelling.
  • Finally, Confluent Co-founder and VP of Product Neha Narkhede moderated a panel of industry influencers in Stream of Consciousness.

  

  And here’s some of the content that stood out to me:

  • Benny Lee and Christopher Arthur of the Commonwealth Bank of Australia continued the trend of sharing long-term enterprise deployment success stories with Kafka in the Enterprise – A Two-Year Journey to Build a Streaming Platform from Scratch.
  • I saw a crowd gathered around Randall Hauch for a least a half-hour after he delivered So You Want to Write a Connector?, his tour of the internals and APIs of Kafka Connect.
  • Nick Dearden tackled the notoriously difficult topics of how to join streams and tables in their various combinations in Zen and the Art of Streaming Joins. This is essential material for mastering KSQL and Kafka Streams.
  • Lyndon Hedderly contributed to our new Business Case track with How to Quantify the Value of Kafka in Your Organization.
  • Community powerhouse Stéphane Maarek failed to pull punches (but succeeded in sharing his knowledge) in Kafka Security 101: Real World Tips.
  • Wanny Morellato of SAP Concur has been running Kafka in production for eight years or so. In Breaking Down a SQL Monolith with Change Tracking, Kafka and KStreams/KSQL, he takes the novel approach of showing how to refactor a conventional SQL query into Kafka Streams and KSQL code.

  

  I hope you can spend some time with the videos. If you’ve got something to share with the community, start writing an outline, because the the call for papers for Kafka Summit NYC 2019 (April 2, 2019) and Kafka Summit London 2019 (May 13-14, 2019) will be opening soon. I hope to see you there!

  Interested in more?

  If you’d like to know more, you can download the Confluent Platform, the leading distribution of Apache Kafka.

  
  
  https://www.confluent.io/blog/kafka-summit-san-francisco-2018-roundup
• 

  Apache Kafka on Kubernetes – Could you?Should you?
  kubernetes apache kafka confluent platform confluent operator helm charts • • sorangutan  

  1
  1
  Posts
  92
  Views

  

  When Confluent launched the Helm Charts and early access program for Confluent Operator, we published a blog post explaining how to easily run Apache Kafka^® on Kubernetes. Since then, we’ve heard from quite a few members of the community question whether running Kafka on Kubernetes is actually a good idea.

  For example, my friend Dylan works for an insurance company and wondered, “…Kubernetes and Kafka are complex distributed systems…I know that Confluent has the expertise to do the management of it; however, what do you think of a normal enterprise dealing with it? Or in other words, just because one could, does that mean we really should?”I completely agree with Dylan that you need to evaluate every technology for fit within both your overall architecture, and the skillset and culture of the organization. Not everything that works in a small Silicon Valley startup will work for a large traditional enterprise.

  With that in mind, there are actual good reasons for running Kafka on Kubernetes within a traditional enterprise company.

  Why run Kafka on Kubernetes?

  First, if you are running most of your other applications and microservices on Kubernetes, it becomes the organizational path of least resistance. This is just like how organizations who standardized on VMs have found it very difficult to allocate physical machines with local disks for Kafka.

  I see situations with larger organizations where deploying Kafka outside of Kubernetes causes significant organizational headache that involves many approvals. When this is the case, I usually say that this isn’t a good hill to die on. It is possible to run Kafka on Kubernetes, so just do it. You’ll get your environment allocated faster and will be able to use your time to do productive work rather than fight an organizational battle.
  And if things go wrong, you’ll get much better service from your internal infrastructure teams, because you’ll be running in an environment that is familiar to them.

  Second, most orgs underestimate how many Kafka clusters they will end up deploying. As the number of use cases grow, you end up with multiple production clusters. And of course there are development environments, testing environments, trying-out-a-new-version environments, blue-green deployment environments and so on.

  Kubernetes, especially with Confluent Operator, does make it easier to deploy and manage new clusters. You could use other deployment tools like Confluent’s Ansible scripts, but they don’t have the built-in provisions for scaling out, monitoring, restarts, upgrades, etc.

  Once you get used to Kubernetes (and it does not take long), you’ll see that Kafka management becomes much easier. It becomes easier to scale up—adding new brokers is a single command or a single line in a configuration file. And it is easier to perform configuration changes, upgrades and restarts on all brokers and all clusters.

  Kafka on Kubernetes can be challenging

  Kafka is a stateful service, and this does make the Kubernetes configuration more complex than it is for stateless microservices. The biggest challenges will happen when configuring storage and network, and you’ll want to make sure both subsystems deliver consistent low latency.

  Kafka on Kubernetes, and other stateful services, require the use of shared storage. Support for local persistent storage on Kubernetes is still in beta and not recommended for production, though now may be a good time to start testing it on a pilot project, as it will likely become GA in few months.

  Unfortunately, many organizations still can’t deliver consistent low latency on their shared storage device. If you want to run Kafka successfully on Kubernetes, you need to make sure the storage team understands the requirements and that together you validate that they are met.

  Kafka also poses a challenge that most stateful services don’t: Brokers are not interchangeable, and clients will need to communicate directly with the broker that contains the lead replica of each partition they produce to or consume from. You can’t place all brokers behind a single load balancer address. You need a way to route messages to specific brokers.

  This isn’t particularly difficult. In the Recommendations for Deploying Apache Kafka on Kubernetes paper, we explained the available options and their tradeoffs. In most cases, the process will require cooperation from the network team.

  The main lesson here is that you can run Kafka on Kubernetes successfully if, and only if, you have the cooperation of skilled storage and network teams. If you don’t have those, you will run into trouble. But the same has always been true—Kafka depends on good infrastructure, and customers who lack good corporate infrastructure run into issues no matter what.

  In addition, I’d recommend against choosing Kafka as the first service to run on Kubernetes. Let the infrastructure team gain experience in deploying, monitoring, updating and troubleshooting stateless services first, such as Kafka Streams applications.

  As mentioned earlier, any single app on Kubernetes doesn’t bring you much benefits. Kubernetes really shines when you use it to manage all your applications and infrastructure. Running Kafka brokers on Kubernetes is most beneficial if the line of business applications are running on Kubernetes too.

  Now some engineers look at the fact that you need to configure persistent volumes on shared storage and headless services with load balancer policies, and see these as “workarounds” and an indication that StatefulSets on Kubernetes are not quite mature yet. I don’t see things this way. In my view, Kafka has specific requirements, and Kubernetes delivers the mechanisms to support Kafka. With version 1.9, StatefulSets are a first-class citizen in the Kubernetes ecosystem.

  Conclusion

  If you use tools like Helm Charts for provisioning and Confluent Operator to make running Kafka on Kubernetes easy, and if you have the support of the storage and network teams, Kafka on Kubernetes is quite easy—in my opinion, easier than any other way of running Kafka.

  You can get started with Kafka on Kubernetes today by checking out the white papers and Helm Charts available online. With Confluent Operator, we are productizing years of Kafka experience with Kubernetes expertise to offer you the best way of using Apache Kafka on Kubernetes.

  Let us know if you are interested, and we may invite you to join our beta program.

  
  
  https://www.confluent.io/blog/apache-kafka-kubernetes-could-you-should-you
• 

  Noise Mapping with KSQL, a Raspberry Pi and a Software-Defined Radio
  ksql stream processing use case use cases streams kafka topic • • sorangutan  

  1
  1
  Posts
  66
  Views

  

  This guest post by Simon Aubury, a data engineer architect from Sydney, Australia, is based on his article Using KSQL, Apache Kafka, a Rasperry Pi and a software defined radio to find the plane that wakes my cat.

  Our new cat, Snowy, is waking early. She is startled by the noise of jets flying over our house. Can I determine which plane is upsetting her by utilizing open source streaming solutions (Apache Kafka^®), KSQL (the streaming SQL engine for Apache Kafka) and a Raspberry Pi? Perhaps a nice dashboard will distract her and let me get a bit more sleep.

  Overview

  Planes to graphs using Kafka and KSQL

  Aircraft determine their position using GPS receivers. An onboard transponder periodically transmits the position along with aircraft identity string, altitude and speed using short radio transmissions. These automatic dependent surveillance-broadcast (ADS-B) transmissions are data packets that can be freely received by ground stations.

  A tiny computer, such as a Raspberry Pi, and a handful of supporting components are all that are required to receive aircraft transponder messages of the planes flying over my house.

  These aircraft transmissions are not coordinated—the messages appear as a jumble of interwoven transmissions. Unraveling these mixed data streams is complex , like trying to understand a conversation when everyone’s talking at the same time at a party. Therefore, I decided to use a combination of Kafka and KSQL to find the plane that’s upsetting my cat.

  

  Here is Snowy looking wide awake.

  ADS-B capture using a Raspberry Pi

  To capture the aircraft transmissions, I used a Raspberry Pi and a RTL2832U ,  which is a USB dongle originally sold to watch digital TV on a computer. On the Pi, I installed dump1090 ,  a program which accesses ADS-B data via the RTL2832U and a small antennae.

  

  Raspberry Pi and RTL2832U as a software-defined radio (SDR)

  ADS-B signals to Kafka topics

  Now that I’ve got a stream of raw ADS-B signals, we need to look at the traffic. The Raspberry Pi is a little underpowered for serious computation, so I need to offload processing to my local Kafka cluster.

  

  Received messages are separated into either a location update message or identification message. A location update would be of the form airframe 7c6db8 is flying at 6,250 feet at location -33.8,151.0. An identification message will be of the form airframe 7c451c is operating flight route QJE1726.

  A small Python script running on the Raspberry Pi separates the incoming ADS-B messages. The Confluent REST Proxy is used to relay producer data from the Raspberry Pi into the location-topic and ident-topic Kafka topics. The proxy provides a RESTful interface to a Kafka cluster, making it easy to produce messages by simply invoking a REST call on the Pi.I wanted to understand what the planes were and what routes they were flying. OpenFlights has a database of airframes which allow the mapping of an airframe International Civil Aviation Organization (ICAO) code, such as 7C6DB8, to an aircraft type—a Boeing 737, in this case. I loaded this mapping into the icao-to-aircraft topic.

  KSQL provides a “SQL engine” that enables real-time data processing against Apache Kafka topics. For example, to look up an airframe code, such as code 7C6DB8, we can query our aircraft topic like this:

  CREATE TABLE icao_to_aircraft WITH (KAFKA_TOPIC='ICAO_TO_AIRCRAFT_REKEY', VALUE_FORMAT='AVRO', KEY='ICAO');
  
  ksql> SELECT manufacturer, aircraft, registration \
  FROM icao_to_aircraft \
  WHERE icao = '7C6DB8';
  Boeing | B738 | VH-VYI

  Similarly, into the callsign-details topic I loaded callsign details (i.e., QFA563 is a Qantas flight from Brisbane to Sydney).

  CREATE TABLE callsign_details WITH (KAFKA_TOPIC='CALLSIGN_DETAILS_REKEY', VALUE_FORMAT='AVRO', KEY='CALLSIGN');
  
  ksql> SELECT operatorname, fromairport, toairport \
  FROM callsign_details \
  WHERE callsign = 'QFA563';
  
  Qantas | Brisbane | Sydney

  Let’s have a peek at the location-topic stream. You can see a steady stream of incoming messages reporting location updates from passing aircraft.

  kafka-avro-console-consumer --bootstrap-server localhost:9092 --property --topic location-topic
  
  {"ico":"7C6DB8","height":"6250","location":"-33.807724,151.091495"}

  The equivalent KSQL syntax is:

  ksql> SELECT TIMESTAMPTOSTRING(rowtime, 'yyyy-MM-dd HH:mm:ss'), \
  ico, height, location \
  FROM location_stream \
  WHERE ico = '7C6DB8';
  
  2018-09-19 07:13:33 | 7C6DB8 | 6250.0 | -33.807724,151.091495

  KSQL:  Crossing the streams…

  The real power of KSQL comes from combining the incoming stream of location data against the reference data topics (see 03_ksql.sq)—that is, adding useful details to the raw data stream. This is very similar to a left join in a traditional database. The result is another Kafka topic—produced without a single line of Java code!

  CREATE STREAM location_and_details_stream AS \
  SELECT l.ico, l.height, l.location, t.aircraft \
  FROM location_stream l \
  LEFT JOIN icao_to_aircraft t ON l.ico = t.icao;

  And you also get a KSQL query. The stream looks like this:

  ksql> SELECT TIMESTAMPTOSTRING(rowtime, 'yy-MM-dd HH:mm:ss') \
  , manufacturer \
  , aircraft \
  , registration \
  , height \
  , location \
  FROM location_and_details_stream;
  18-09-27 09:53:28 | Boeing | B738 | VH-YIA | 7225 | -33.821,151.052
  18-09-27 09:53:31 | Boeing | B738 | VH-YIA | 7375 | -33.819,151.049
  18-09-27 09:53:32 | Boeing | B738 | VH-YIA | 7425 | -33.818,151.048

  Equally, we can combine the incoming callsign identity stream against the static callsign_details topic:

  CREATE STREAM ident_callsign_stream AS \
  SELECT i.ico \
  , c.operatorname \
  , c.callsign \
  , c.fromairport \
  , c.toairport \
  FROM ident_stream i \
  LEFT JOIN callsign_details c ON i.indentification = c.callsign;
  
  ksql> SELECT TIMESTAMPTOSTRING(rowtime, 'yy-MM-dd HH:mm:ss') \
  , operatorname \
  , callsign \
  , fromairport \
  , toairport \
  FROM ident_callsign_stream ;
  18-09-27 13:33:19 | Qantas | QFA926 | Sydney | Cairns
  18-09-27 13:44:11 | China Eastern | CES777 | Kunming | Sydney
  18-09-27 14:00:54 | Air New Zealand | ANZ110 | Sydney | Auckland

  We now have two enriched topics:

  1. location_and_details_stream, which provides a stream of aircraft location and speed updates
  2. ident_callsign_stream, which describes the details of flights, including carrier and destination

  With these constantly updating topics, we can make some pretty dashboards. I used Kafka Connect to pump the Kafka topics that KSQL populates into Elasticsearch (full scripts).

  Kibana dashboard

  Here’s a sample of dashboards displaying aircraft location on a map. In addition, you can see a bar chart of manufactures, altitude line plot and destination word-cloud. A heat map shows areas where aircraft are concentrated, where we might hear increased noise levels.

  

  Kibana display

  Back to the cat

  

  What I woke up to this morning…

  Today, my cat woke me at a little past 6:00 a.m. Can I use KSQL to find a low flying aircraft (below 3,500 feet) around this time?

  select timestamptostring(rowtime, 'yyyy-MM-dd HH:mm:ss')
  , manufacturer
  , aircraft
  , registration
  , height
  from location_and_details_stream
  where height < 3500 and rowtime > stringtotimestamp('18-09-27 06:10', 'yy-MM-dd HH:mm') and rowtime < stringtotimestamp('18-09-27 06:20', 'yy-MM-dd HH:mm');
  
  2018-09-27 06:15:39 | Airbus | A388 | A6-EOD | 2100.0
  2018-09-27 06:15:58 | Airbus | A388 | A6-EOD | 3050.0

  Terrific, I can locate a plane over my house at 6:15 a.m. It’s an Airbus A380 (which is a huge plane) on a flight to Dubai that’s waking my cat.

  Developing the stream processing with KSQL allowed for some rapid weekend development. Plus, KSQL allows for quick discovery of interesting data events. My cat Snowy, however, might take a bit of convincing.

  Ready to try it?

  Got a similar pet problem, or just curious about how Kafka and KSQL work together? You can find a combined Docker setup, helper images, KSQL and Elastic setup on GitHub.

  If you’re interested in what KSQL can do:

  You can download the Confluent Platform, the leading distribution of Apache Kafka, to get started with KSQL.

  
  
  https://www.confluent.io/blog/noise-mapping-ksql-raspberry-pi-software-defined-radio
• 

  Event Driven 2.0: Data as a Service
  events microservices architecture streaming platform • • sorangutan  

  1
  1
  Posts
  257
  Views

  

  In the future, data will be as automated and self-service as infrastructure is today. You’ll open a console that lists the data available in your company; define the pieces you need, the format you want and how you would like it joined together; spin up a new endpoint: a database, cache, microservice or serverless function and off you’ll go.

  These are the event-driven architectures of the modern age—but ones where messaging is far more than a simple pipe that connects systems together. Part messaging system, part distributed database, streaming systems let you store, join, aggregate and reform data from across a company, before pushing it wherever it is needed, be it a hulking data warehouse or a tiny serverless function.

  When combined with public and private clouds, where infrastructure can be provisioned dynamically, the result makes data entirely self- service.

  Many companies have implemented some version of this future. The different approaches they take can be split into four broad patterns, and the companies and projects we see typically adopt them one at a time:

  1. Global event streaming platform
  2. Central event store
  3. Event-first and event streaming applications
  4. Automated data provisioning

  No one company that we know of has mastered them all, but all these categories exist in production in one form or another.

  1. Global event streaming platform

  This is the easiest to grok, as it is similar to the enterprise messaging patterns of old. The organization moves to an event-driven approach with core datasets (those shared between applications, such as orders, customers, payments, accounts, trades, etc.) flowing through an event streaming platform like Apache Kafka^®.

  These replace legacy point-to-point communications with a single piece of infrastructure that lets applications operate across different geographies or cloud providers, at scale and in real time.

  So, a company might run a legacy mainframe in San Francisco, have regional offices in Cape Town and London, as well as have highly available microservices running across AWS and GCP—all connected with the same event backbone. More extreme use cases include connecting ships via satellite or automobiles via mobile.

  Companies implement this pattern in almost every industry. Examples are Netflix, Audi, Salesforce, HomeAway, ING and RBC, to name a few.

  

  2. Central event store

  Streaming platforms can cache events for some defined period of time or store them indefinitely, creating a kind or organizational ledger or event store.

  Some companies use this pattern to drive retrospective analysis, for example training the machine learning models used in fraud detection or rewinding time in Formula 1 post-race analysis. Others apply the pattern across many teams.

  This lets new applications be built without requiring source systems to republish prior events, a property that is particularly useful for datasets that are hard to replay from their original source, such as mainframe, external or legacy systems.

  Some organizations hold all their data in Kafka. The pattern is termed a forward event cache, event streaming a source of truth, the kappa architecture or simply event sourcing.

  Finally, event storage is required for stateful stream processing, which is often applied to create enriched, self-sufficient events from many different data sources. This could be enriching orders with customer or account information, for instance.

  Enriched events are easier to consume from microservices or FaaS implementations because they provide all the data the service needs. They can also be used to provide a denormalized input for databases. The stream processors that perform these enrichments need event storage to hold the data that backs tabular operations (join to customers, accounts, etc.).

  

  3. Event-first and event streaming applications

  Most traditional applications work by importing datasets from different places into their database (e.g., ETL), where it can be cleansed, joined, filtered and aggregated.

  For applications that create reports, dashboards, online services, etc., this remains the best choice, but for business processing, it is often more efficient to skip the database step by pushing real-time events directly into microservices or serverless functions.

  In such approaches, stream processors like Kafka Streams or KSQL perform the data manipulation that the database did in the traditional approach by cleansing, joining, filtering and aggregating event streams before they are pushed into the microservice or FaaS.

  As an example, consider a limit checking service that joins orders and payments together using a stream processor like KSQL, extracting the relevant records/fields and passing them into a microservice or function as a service that checks the limit—a workflow in which no database is used at all.

  Due to their event-driven nature, such systems are more responsive. They are also typically simpler and faster to build, as there is less infrastructure and data to maintain, and the toolset naturally handles asynchronously connected environments.

  Richer examples incorporate stream analysis directly, such as detecting anomalous behavior in credit card payments or optimizing energy delivery in a smart power grid. Such systems often exist as chains, where the stages separate stateful and stateless operations, can scale independently and leverage transactional guarantees for correctness.

  We see applications of this type crop up across many industries: finance, gaming, retail, IoT, etc., spanning both offline and online use cases.

  

  4. Automated data provisioning

  This final pattern is the culmination of the others, blended with PaaS/serverless implementations to make data provisioning entirely self-service.

  A user defines the data they require (real time or historical), what form it should take and where it should be landed, whether in a database, distributed cache, microservice, FaaS or wherever. (Typically, this is used in association with a central repository of schemas which is made discoverable.)

  The system provisions the infrastructure, pre-populates it where necessary and manages the event flow. A stream processor filters, manipulates and buffers the various shared data streams, moulding them to the user’s specification.

  So, a finance user doing risk analysis might spin up a new Elasticsearch instance pre-populated with three months’ trades, risk results and books. Or, a retail company might join real-time orders, payments and customer data, and push them into a microservice or FaaS that sends the customer a payment confirmation.

  As organizations move to public and private clouds, the dynamic nature of cloud-based infrastructure makes this pattern increasingly practical, leading to systemic benefits. New projects, environments or experimentation can be quickly initiated.

  Because datasets are cached or stored in the messaging system, users are encouraged to take only the data they need at a point in time (unlike traditional messaging where there is a tendency to consume and hoard whole datasets in case they are needed again later). This minimizes friction between teams and keeps applications close to a single, shared source of truth.

  Few organizations that we know of have fully achieved this level of automation, but core elements of this pattern are used in production at several of our customers in the finance, retail and internet domains, both on premises and in the cloud.

  

  Event driven 2.0: An evolution and a new beginning

  Event-driven architectures have evolved naturally over the years. Originally, they only did message passing: notification and state transfer applied with traditional messaging systems.

  Later, enterprise service buses embellished these with richer out-of-the-box connectivity and better centralized control. Centralized control turned out to be a mixed blessing, as the standardization it provided often made it harder for teams to progress.

  More recently, storage patterns like event sourcing and CQRS have become popular, as discussed by Martin Fowler in his article What do you mean by “Event Driven”?

  The four patterns that I’ve described all build on this foundation, but today’s modern event streaming systems allow us to go even further by unifying events, storage and processing into a single platform. This unification is important, because these systems aren’t databases which lock data up in a single place; they aren’t messaging systems, where data is fleeting and transitory. They sit somewhere in between.

  By striking a balance between these two traditional categories, companies have been able to become globally connected across regions and across clouds, with data—their most precious commodity—available as a service, whether that means pushing it into a database, cache, machine learning model, microservice or serverless function.

  So, in summary:

  • Broadcast events
  • Cache shared datasets in the log and make them discoverable.
  • Let users manipulate event streams directly (e.g., with a streaming engine like KSQL)
  • Drive simple microservices or FaaS, or create use-case-specific views in a database of your choice

  Find out more

  To find out more about event-driven systems and how to implement them:

  • Download the free O’Reilly book Designing Event-Driven Systems
  • Watch Confluent CEO Jay Kreps explain event-driven architectures
  • Read the Gartner report Top 10 Strategic Technology Trends for 2018: Event-Driven Model

  
  
  https://www.confluent.io/blog/event-driven-2-0-data-service
• 

  KSQL Recipes Available Now in the Stream Processing Cookbook
  ksql stream processing tutorial • • sorangutan  

  1
  1
  Posts
  113
  Views

  

  For those of you who are hungry for more stream processing, we are pleased to share the recent release of Confluent’s Stream Processing Cookbook, which features short and tasteful KSQL recipes that help you solve specific, domain-focused problems using KSQL.

  Organized according to various use cases (e.g., partitioning, streaming ETL, anomaly detection, data wrangling), KSQL recipes provide easy directions that you can follow to begin—or continue!—putting KSQL to use.

  KSQL is the streaming SQL engine for Apache Kafka^®, and is complementary to the Kafka Streams API. As KSQL and Kafka Streams continue growing in their adoption and support within the stream processing world and Kafka ecosystem, in parallel with increasing investments in evolving streaming applications, these recipes benefit both developers and the business alike by providing pre-baked KSQL techniques that involve little modification, regardless of whether or not you have a Java background.

  Why use KSQL recipes?

  KSQL recipes are designed to help people build event-driven, real-time systems. They are submitted by Confluent specialists and community members, and therefore serve as a place to collaborate, share and inspire one another with KSQL applications, broadening the approach of streaming application development.

  KSQL recipes serve three main purposes:

  1. Making it easier to get started: Instead of starting from scratch, KSQL recipes present an established set of common technical patterns that provide a good starting point, especially for first-time users.
  2. Providing a source of inspiration: The cookbook is a place where you can see how others have solved problems that you are tackling yourself. As you browse, it may inspire you to either use their starting point or branch off and do something completely different. The recipes also offer additional ideas that are drawn from practical scenarios, which you may not have been aware of before.
  3. Encouraging collaboration: If you have created a KSQL code snippet that you would like to share, this is the place to share it! KSQL recipes do include contributions from the community, and we invite you to add your own unique recipe to the cookbook. To do so, simply email [email protected]. But why would you share the snippet, you might ask? Well, others may use it, enhance it and create new applications with your original code, which may eventually lead to the development of new recipes that can benefit you in the future too. (Plus, if it’s popular, maybe even fame!)

  How can I use KSQL recipes?

  The recipes show you how to use KSQL in different ways. Although you have the option to read and follow the recipes as they are, you can also think about how to use the same code patterns but apply them to other use cases.

  As an example, a very commonly used KSQL recipe is Processing Syslog Data. This general code pattern can apply to a variety of use cases, including fraud detection, network traffic, signal anomaly or propensity analysis. All of these use cases can leverage KSQL, as they use a similar code base despite deviating based on the specified implementation of the recipe.

  In addition, building, testing and modifying applications becomes faster with the recipes. Because they help with getting started, you can quickly build in customizations and specifics, run the application, examine the output and iterate, all without the help of a Kafka operator.

  This may seem like a marginal advantage, but with the number of and pace at which streaming applications and transformations are being built, all of these cycles are significant. The need to have a Kafka administrator results in a potential bottleneck, which slows the entire development experience intended to move toward agility and defeats the purpose of KSQL.

  Check out the Stream Processing Cookbook

  The Stream Processing Cookbook is now available for you to peruse. If you’re interested, you can browse the KSQL recipes for handy tutorials and examples to follow, or contribute a KSQL recipe by emailing [email protected].

  Let’s get cooking!

  
  
  https://www.confluent.io/blog/stream-processing-cookbook-featuring-ksql-recipes
• 

  Enhance Security with Apache Kafka 2.0 and Confluent Platform 5.0
  security apache kafka confluent platform confluent control center kafka connect • • sorangutan  

  1
  1
  Posts
  91
  Views

  

  As customers across verticals like finance, healthcare, state and local government and education adopt Confluent Platform for mission-critical data, security becomes more and more important. In the latest release of our streaming platform, Confluent Platform 5.0, we introduced several security enhancements to help you solve your most difficult security problems. In this article, I will provide deeper technical analysis about the most important security enhancements that are part of the Confluent Platform 5.0 release, including those from Apache Kafka^® 2.0.

  What you’ve been asking for

  Over the past several quarters, we have made major security enhancements to Confluent Platform, which have helped many of you safeguard your business-critical applications. With the latest release, we increased the robustness of our security feature set to help with:

  • Using standard and central directory services like Active Directory (AD)/Lightweight Directory Access Protocol (LDAP)
  • Simplifying the management of access control lists (ACLs)
  • Proactive management and monitoring of security configurations to address the gaps as soon as possible

  The following new security features are available in both Confluent Platform 5.0 and Apache Kafka 2.0:

  • Support for ACL-prefixed wildcards to simplify the management of access control
  • Kafka Connect password protection with support for externalizing secrets (to “secrets stores,” etc., like Hashicorp Vault)

  The following security features are available only in Confluent Platform 5.0:

  • AD/LDAP group support
  • Feature access controls in Confluent Control Center
  • Viewing of broker configurations in Confluent Control Center, including differences in security configurations between brokers

  Let’s walk through each of these enhancements in detail.

  AD/LDAP group support in Confluent Platform 5.0

  The majority of enterprises standardize on AD for identity-related services. Most people who use directory services would also like to be able to use groups for configuring access controls. In the latest release, we added the capability to use AD and LDAP groups for just this purpose. This simplifies access control management, requiring fewer rules for managing access to groups or organizations. The benefits are:

  • You can manage group ACLs with existing tools and APIs, including kafka-acls and AdminClient.
  • A DENY rule for any of the groups that a principal belongs to will deny access to a resource.
  • An ALLOW rule for any of a principal’s groups will allow access if no DENY rule matches the resource.
  • Users and groups are refreshed periodically, as defined by ldap.authorizer.refresh.interval.ms, enabling the user and group information to keep up to date with the directory server.
  • You can limit the users or groups returned from the LDAP server using the ldap.authorizer.group.search.filter configuration parameter. If you have many users or groups in AD/LDAP, this can be a lifesaver.
  • If you’re using AD, we support authentication using Kerberos and authorization using groups.

  You can learn more by checking out the Kafka LDAP Authorizer documentation.

  Finding broker security configuration mismatches in Confluent Platform 5.0

  Confluent Control Center now gives you the ability to compare the configuration of different brokers in a Kafka cluster and identify any differences between them. This is a great way to see if you have misconfigured brokers that could be contributing to security vulnerabilities.

  

  Feature access controls in Control Center

  Enterprises need to ensure that end users cannot access sensitive data via Control Center for security and compliance reasons. In order to let administrators control application-wide access to features that reveal topic data, we added the capability to control access to topic inspection, schemas and KSQL queries. When you restrict access to one of these features through the configuration file, Control Center’s user interface will reflect this change upon startup.

  Simplify ACL management with support for prefixed ACLs

  ACLs do a great job of controlling access to resources like topics and consumer groups, but they can sometimes lack flexibility like any other kind of sophisticated pattern matching. A resource name in an ACL definition is either a complete resource name or a wildcard that matches everything—no middle ground exists. Oftentimes, you prefer more flexibility in setting up ACLs so you don’t end up with a huge list of them. In Confluent Platform 5.0 and Apache Kafka 2.0, we extended the concept of wildcard ACL (*) to support prefixed ACLs. For example:

  • FinanceUser has access to all topics that start with Finance_
  • UserN has access to all consumer groups that start with FraudDetection_

  Externalizing secrets for Kafka Connect configurations

  Today, Kafka Connect keeps configuration information in clear text, including source and sink credentials. This is not ideal from a security standpoint. At the same time, there are third-party secret stores (like HashiCorp Vault) that provide the ability to protect these passwords at rest and in flight, while also maintaining audit logs, versioning and providing high availability.

  Apache Kafka 2.0 and Confluent Platform 5.0 include a new framework in Kafka Connect that lets you integrate your preferred secret store and then use placeholders for secrets in the connector configurations. Kafka Connect will keep the placeholders in its stored configurations and in all REST requests and responses. It will automatically use your plugin to resolve the placeholders into secrets immediately before the configuration is passed to the connector upon startup.

  Please note that at this point we do not provide built-in secret store plugins—this feature introduces the framework, not the plugins themselves—but you are now able to implement your own.

  Streamlining the management of ACLs for KSQL and Kafka Streams

  All of the enhancements we’ve made across the platform in 5.0 have enabled us to make further security improvements in KSQL and Kafka Streams applications.

  The combination of ACL wildcards, support for AD/LDAP groups and the new configuration setup simplify the security configuration for KSQL and Kafka Streams, which is very helpful particularly for teams who are deploying and operating KSQL and Kafka Streams applications in secured, multi-tenant production environments.

  This enables developers faster time to production by streamlining the process of setting access control on source, internal and sink topics.

  Give us feedback

  As always, we are happy to hear your feedback. Please post your questions and suggestions to the public Confluent Platform mailing list, join our community Slack channel or contact us directly!

  Interested in more?

  If you’d like to know more, here are some resources for you:

  • You can download the Confluent Platform
  • Read our security documentation
  • Confluent Professional Services offers advice and help with deployments
  • Confluent Training can get your team ready for development and deployment of the Confluent Platform

   

  
  
  https://www.confluent.io/blog/enhance-security-apache-kafka-2-0-confluent-platform-5-0
• 

  Troubleshooting KSQL – Part 2: What’s Happening Under the Covers?
  ksql stream processing troubleshooting confluent control center jmx • • sorangutan  

  1
  1
  Posts
  93
  Views

  

  Previously in part 1, we saw how to troubleshoot one of the most common issues that people have with KSQL—queries that don’t return data even though data is expected. Here, we’ll dig deeper into the internals for more advanced troubleshooting and better understand how it all works.

  You can use this article for reference, or follow along with the code examples to try them out as we go. Using Docker and Docker Compose, we can easily provision an environment in which to explore and try out the different techniques and tools. The environment includes a data generator for a continuous stream of events into an Apache Kafka^® topic that we will use for testing. All the necessary code is available on GitHub.

  Let’s dive in and start exploring what to do when things aren’t working…

  How many messages have been processed by a KSQL query?

  In KSQL you can populate Kafka topics with the results of a query using the CREATE STREAM…AS SELECT syntax:

  ksql> CREATE STREAM GOOD_IOS_RATINGS AS \
          SELECT * FROM RATINGS WHERE STARS >= 4 \
                                 AND CHANNEL='iOS';

  Because KSQL queries are continuous, this means that we’ve just written and executed an application. It takes the inbound data, filters it for a condition and writes any matches to the target topic.

  What does any self-respecting application need? Metrics! We need to know how many messages have been processed, when the last message was processed and so on.

  The simplest option for gathering these metrics comes from within KSQL itself, using the same DESCRIBE EXTENDED command that we saw before:

  ksql> DESCRIBE EXTENDED GOOD_RATINGS;
  [...]
  Local runtime statistics
  ------------------------
  messages-per-sec:      1.10 total-messages:     2898 last-message: 9/17/18 1:48:47 PM UTC
   failed-messages:         0 failed-messages-per-sec:         0 last-failed: n/a
  (Statistics of the local KSQL server interaction with the Kafka topic GOOD_RATINGS)
  ksql>

  Also note that an increasing failed-messages count is not a good sign for the health of your query. It could be caused by serialization errors, as discussed in part 1. The metrics shown here are local to the server on which the DESCRIBE statement executes. The documentation includes definitions of all the metrics.

  What’s happening under the covers?

  To dig deeper into the execution of queries, let’s start by listing the queries that are running:

  ksql> SHOW QUERIES;
  
   Query ID                | Kafka Topic | Query String
  --------------------------------------------------------------------------------------------------------------------------------------------------------------------------
   CSAS_GOOD_IOS_RATINGS_0 | GOOD_IOS_RATINGS | CREATE STREAM GOOD_IOS_RATINGS AS     SELECT * FROM RATINGS WHERE STARS >= 4 AND CHANNEL='iOS';
  --------------------------------------------------------------------------------------------------------------------------------------------------------------------------

  Just the one query is running, populating GOOD_IOS_RATINGS, as defined in the CSAS statement we ran above. The query is called CSAS_GOOD_IOS_RATINGS_0 (which is non-deterministic).

  We can examine the query itself and how KSQL is going to perform the transformation we’ve asked it to through the explain plan—the same thing as you get in a RDBMS. You can access it by using the EXPLAIN command:

  ksql> EXPLAIN CSAS_GOOD_IOS_RATINGS_0;
  
  [...]
  
  Execution plan
  --------------
  > [ SINK ] Schema: [ROWTIME : BIGINT, ROWKEY : VARCHAR, RATING_ID : BIGINT, USER_ID : BIGINT, STARS : INT, ROUTE_ID : BIGINT, RATING_TIME : BIGINT, CHANNEL : VARCHAR, MESSAGE : VARCHAR].
                   > [ PROJECT ] Schema: [ROWTIME : BIGINT, ROWKEY : VARCHAR, RATING_ID : BIGINT, USER_ID : BIGINT, STARS : INT, ROUTE_ID : BIGINT, RATING_TIME : BIGINT, CHANNEL : VARCHAR, MESSAGE : VARCHAR].
                                   > [ FILTER ] Schema: [RATINGS.ROWTIME : BIGINT, RATINGS.ROWKEY : VARCHAR, RATINGS.RATING_ID : BIGINT, RATINGS.USER_ID : BIGINT, RATINGS.STARS : INT, RATINGS.ROUTE_ID : BIGINT, RATINGS.RATING_TIME : BIGINT, RATINGS.CHANNEL : VARCHAR, RATINGS.MESSAGE : VARCHAR].
                                                   > [ SOURCE ] Schema: [RATINGS.ROWTIME : BIGINT, RATINGS.ROWKEY : VARCHAR, RATINGS.RATING_ID : BIGINT, RATINGS.USER_ID : BIGINT, RATINGS.STARS : INT, RATINGS.ROUTE_ID : BIGINT, RATINGS.RATING_TIME : BIGINT, RATINGS.CHANNEL : VARCHAR, RATINGS.MESSAGE : VARCHAR].

  Because KSQL is built on Kafka Streams and executes queries using it, the EXPLAIN command can also tell you the topology that Kafka Streams will use:

  <ksql> EXPLAIN CSAS_GOOD_IOS_RATINGS_0;
  
  [...]
  
  Processing topology
  -------------------
  Topologies:
     Sub-topology: 0
      Source: KSTREAM-SOURCE-0000000000 (topics: [ratings])
        --> KSTREAM-MAPVALUES-0000000001
      Processor: KSTREAM-MAPVALUES-0000000001 (stores: [])
        --> KSTREAM-TRANSFORMVALUES-0000000002
        <-- KSTREAM-SOURCE-0000000000
      Processor: KSTREAM-TRANSFORMVALUES-0000000002 (stores: [])
        --> KSTREAM-FILTER-0000000003
        <-- KSTREAM-MAPVALUES-0000000001
      Processor: KSTREAM-FILTER-0000000003 (stores: [])
        --> KSTREAM-MAPVALUES-0000000004
        <-- KSTREAM-TRANSFORMVALUES-0000000002
      Processor: KSTREAM-MAPVALUES-0000000004 (stores: [])
        --> KSTREAM-MAPVALUES-0000000005
        <-- KSTREAM-FILTER-0000000003
      Processor: KSTREAM-MAPVALUES-0000000005 (stores: [])
        --> KSTREAM-SINK-0000000006
        <-- KSTREAM-MAPVALUES-0000000004
      Sink: KSTREAM-SINK-0000000006 (topic: GOOD_IOS_RATINGS)
        <-- KSTREAM-MAPVALUES-0000000005

  Taking the query name, you can even go and poke around the KSQL server log, and see the Kafka Streams applications firing up with the query name as part of their client ID:

  ksql-server_1      | [2018-09-19 21:05:40,625] INFO stream-thread [_confluent-ksql-confluent_rmoff_01query_CSAS_GOOD_IOS_RATINGS_0-c36ebad2-f969-40e1-9b59-757305cf3b61-StreamThread-5] State transition from CREATED to RUNNING (org.apache.kafka.streams.processor.internals.StreamThread:209)

  The KSQL server log is where you’d head to if you’re suspecting problems with your queries that you can’t diagnose through the KSQL CLI itself. To locate the server log, see here.

  Digging deeper: KSQL troubleshooting with Confluent Control Center

  Part of Confluent Enterprise, Confluent Control Center gives you powerful monitoring, management and alerting capabilities for your Apache Kafka environment. It provides a KSQL editor for building and exploring KSQL objects:

  

  It also has a powerful streams monitoring capability.

  With streams monitoring, you can answer these important questions:

  • What’s the message throughput rate of the pipeline?
  • Were all messages that were written to Kafka consumed by the downstream application,  including KSQL?
  • Were any messages consumed more than once?
  • What was the latency of each consumer?

  

  You can drill down into individual topics and consumers to inspect their particular behavior. Here, we can see that there are two Kafka Streams threads, only one of which (StreamThread-2) is consuming data. The reason for that is the source topic has a single partition:

  

  If we re-partition the topic to four partitions, the Kafka Streams/KSQL task then consumes this over four threads:

  Confluent Control Center can also show you when consumers are over-consuming, which can occur when messages are being processed more than once—or as the case is in this screenshot, the required monitoring interceptors haven’t been set up on the producer.

  

  Digging deeper: KSQL troubleshooting with JMX

  Confluent Control Center is very cool for inspecting the flow of data in topics and behavior of producers and consumers, but what if we peeked inside how those producers and consumers are actually behaving? KSQL, as with other components in the Apache Kafka ecosystem, exposes a wealth of metrics through JMX. You can access these on an ad hoc basis through something like jmxterm or JConsole, as seen here:

  

  Even more useful is persisting them to a data store, such as InfluxDB, for subsequent analysis. This analysis can be done through Grafana, for example:

  

  You can see messages being consumed by KSQL—all data from the ratings topic—and produced by it to the GOOD_IOS_RATINGS stream with messages matching the STARS >= 4 AND CHANNEL='iOS' criteria. Just by looking at it, you can determine that roughly 15 percent (0.26 divided by 1.96) of the messages are passing through from the ratings source to the GOOD_IOS_RATINGS target:

  

  The spike at 12:31 coincides with the example I ran in part 1 of trying to read messages with the wrong serialization format defined. Conveniently enough, there’s also a JMX metric we can track for errors:

  The spike in error-rate as well as the increase in num-active-queries likewise make sense, because the additional query was running against the invalid stream at the time on top of the fact that there was one already running against ratings.

  You can dig down into the underlying Kafka Streams metrics:

  

  Kafka Streams itself is built on Kafka, and you can drill down to the underlying Kafka producer and consumer metrics too:

  

  

  If you want to try this out for yourself and explore the JMX metrics, the complete code samples are available on GitHub.

  For details of the specific metrics, see:

  • Kafka Streams documentation
  • Apache Kafka documentation

  Need more help?

  Still stuck and need more help? Here are several places to turn:

  • Get support for KSQL as part of Confluent Enterprise
  • Seek community support for KSQL:
    • Ask questions in the Confluent Community Slack #ksql channel
    • Search for similar issues on GitHub
    • Raise a new issue on GitHub if it doesn’t already exist
  • Follow KSQL recipes for tutorials and recommended deployment scenarios

  Other articles in this series:

  • Troubleshooting KSQL – Part 1: Why Isn’t My Query Returning Data?

  
  
  https://www.confluent.io/blog/troubleshooting-ksql-part-2
• 

  Is Event Streaming the New Big Thing for Finance?
  events streaming use cases event streams financial services • • sorangutan  

  1
  1
  Posts
  117
  Views

  

  While Silicon Valley is always awash with new technology buzzwords, one that has become increasingly dominant in recent years is stream processing: a type of software designed to transport, process and react in real time to the massive streams of event data at the heart of large internet companies. While it isn’t entirely surprising that a company like Netflix—responsible for 37 percent of U.S. downstream internet traffic—would be interested in technology that processes data as it moves, what is surprising is the uptake in traditional markets like finance, which don’t typically generate the trillions-of-events-per-day workloads seen at giant technology firms. In fact, this uptake demonstrates a subtle and intriguing shift in the way that companies are reshaping their IT functions. In much the same way that the internet transformed how people are connected, today applications are revolutionizing how big companies operate, interact and, most importantly, share data.

  The digital customer experience

  When a customer interacts with their bank to make a payment, withdraw cash or trade stock, that single action or event triggers a slew of activity: Accounts are updated; postings hit ledgers; fraud checks are run; risk is recalculated and settlement processes are initiated. So the value of these events to the organization is highest in the moment they are created, the moment in which a broad set of operations all need to happen at once. The more widely and immediately available these events are, the easier it is for a company to adapt and innovate.

  A number of upstart financial services firms were quick to notice the importance of being event driven. The UK-based upstart bank Monzo—which focuses heavily on a mobile-first, immersive user experience—makes for a good example. When customers buy goods with their Monzo card, their mobile application alerts them of the transaction often before the receipt has finished printing at the checkout, complete with details of how much they spent and where they spent it. The bank performs real-time categorization on purchases so that users know what they are spending their money on, and the associated mobile notification even includes a purchase-related emoji: a coffee cup for Starbucks, a T-shirt for apparel, etc.

  A cynic might view this as a gimmick, but Monzo’s thriving digital ecosystem is more likely reflective of a substantial shift in customer needs. Customers no longer drive to a branch and talk to their account manager face to face. Instead, they expect a rich and personal digital experience—one that makes them feel in touch with their finances and partnered with a bank that knows what they, the customer, is doing.

  But knowing your customer isn’t simply about collecting their digital footprint—it’s about interacting with it and them in real time. Financial services companies like Funding Circle and Monzo achieve this by repurposing the same streaming technologies used by technology giants like Netflix and LinkedIn. Funding Circle use stream processing tools to, as they put it, “turn the database inside out,” a metaphor that illustrates how the technology lets them manipulate event streams as they move through the organization. In Monzo’s case, as soon as a payment is acquired and the balance updates, the payment event is stored in a streaming platform, pushing those events (or any previous ones) into a wide variety of separate IT services: categorization, fraud detection, notification, spending patterns and, of course, identifying the correct and all-important emoji.

  But being able to adapt a stream of business events with a stream processor isn’t just about upstarts leveraging coffee cup GIFs and immersive user experiences. More traditional institutions like ING, Royal Bank of Canada (RBC) and Nordea have made the shift to streaming systems too. What makes their approach interesting is their interpretation of the event stream, not as some abstract piece of technology infrastructure, but rather as a mechanism for modeling the business itself. All financial institutions have an intrinsic flow that is punctuated with real-world events, be it the processing of a mortgage application, a payment or the settlement of an interest rate swap. By rethinking their business not as a set of discrete applications or silos but rather as an evolving flow of business events, they have created a more streamlined and reactive IT service that better models how the business actually works in a holistic sense.

  The introduction of the MiFID II regulation, which requires institutions to report trading activity within a minute of execution, presented a useful litmus test. This was notoriously difficult for many banks to adhere to due to their siloed, fragmented and batch-oriented IT systems. But for the banks that had already installed real-time event streaming and plumbed it across all their product-aligned silos, the task was far simpler.

  From data warehousing to event streaming

  While streaming technologies change the way organizations communicate internally, they also change the way they store and share data. Today’s streaming systems have as much in common with distributed databases as they do with traditional messaging systems—they store data long term, support streaming SQL queries and scale almost indefinitely, leading to an entirely new category of data technology.

  This is the most recent of a long line of data technology step-changes that finance firms have adopted, each correcting a problem left by the last. Data warehousing lets companies create a single joined-up, high-quality view from a plethora of databases dotted around the organization. Big data technologies refined this approach, but the implementation patterns remained broadly the same: large semi-static data repositories positioned in the aft of the company and crawled daily or monthly to produce management or analytic insight.

  While this is acceptable for periodic reports, the role of data in the modern bank has expanded significantly. Finance firms today use machine learning models to automate decisions off data sourced from various corners of the company and create new applications and “microservices” that provide collaborative capabilities enterprise-wide, cutting across silos and unifying customer experiences. All these initiatives require rich, high-quality, real-time datasets to keep the time to market of these initiatives low.

  To provide these efficiency gains, a number of financial institutions both big and small have repositioned their companywide datasets forward in the organization, shifting their focus away from collecting a high-quality dataset in one place, to delivering to many systems company-wide. For the companies that do this well, data becomes a self-service commodity.

  In one large bank, starting a new project is as simple as logging into a portal and picking the relevant events: trades, books, cash flows, etc., and pushing them into an application or database located on their private cloud. This works across datacenters, geographies and in the future, cloud providers. So, while streaming systems bring real-time connectivity, they also provide unified data enablement, making data a self-service commodity available on a global scale.

  

  Unlocking agility from the assets within

  While streaming lets banks like ING or RBC be more nimble and respond instantaneously to their customers, a potentially greater benefit comes from an increased ability to adapt and evolve the investments they have already made.

  For the last two decades, IT culture has shifted towards more agile approaches for building and delivering software. Waterfall processes, where requirements are locked down before software development commences, have been replaced with more iterative approaches where new software is developed in short, often week-long cycles, from where it is visible, testable and available for feedback. These methods are beneficial because they reassign the notion of value away from fixed goals (i.e., software that does ABC delivered on date XYZ) into process goals (i.e., a process that builds software incrementally), letting customers provide fast feedback and allowing the business to redefine and optimize the end product as the world around it changes.

  This mindset tends to work well for individual software projects, but for most established enterprises—already backed by many billions of dollars of technology investment, built up over decades of existence—embracing the same kind of agility at an organizational level seems impossible. In fact, the boardrooms of financial services companies are far more likely to be graced with proposals for reinvention through the latest shiny new software platform than the less bonus-optimized steps needed to move an aging technology stack incrementally into the future.

  But one of the greatest ills that face the IT industry today is attributing the dysfunction of an IT organization to “dysfunctional software,” when “dysfunctional data” is far more culpable. A company may have best-in-class software engineers, but if the first thing they have to do when they start a new project is identify, collect, import and translate a swathe of flawed and hard-to-access corporate datasets, it is reasonable to expect the same painful release schedules seen in the days of waterfall, not to mention the likelihood that those best-in-class software engineers will migrate elsewhere.

  Companies that take the stream processing route observe a somewhat different dynamic, because the event streams breed agility. These platforms come with tools that transform static databases into event streams, unlocking data hidden deep inside legacy systems and connecting it directly to applications company-wide. For companies like RBC, this means pulling data out of their mainframe and making it available to any project that needs it both immediately as well as retrospectively. The insight is simple but powerful: If the data is always available, then the organization is always free to evolve.

  The implication for the embattled incumbents of the finance industry is that their best strategy may not be to beat the upstarts at their own game. Instead, they are better off unlocking the hidden potential of the assets that they already own. As Mike Krolnik, RBC’s head of engineering for enterprise cloud, put it: “We needed a way to rescue data off of these accumulated assets, including the mainframe, in a cloud native, microservice-based fashion.” This means rethinking IT architectures, not as a collection of independent islands that feed some far-off data warehouse but as a densely connected organism, more like a central nervous system that collects and connects all company data and moves it in real time. So, while buzzword bingo might point to a future of internet velocity datasets, the more valuable benefits of streaming platforms lie in the subtler, systemic effects of denser data connectivity. This may well pass our data-obsessed technology community by, but the wise money is on the long game, and the long game is won with value.

  Interested in more?

  • Read the white paper Top 3 Use Cases for Real Time Streams in Financial Services Architectures
  • Watch our video on digital replatforming in the financial services industry
  • Learn about real-time streams for growth, regulation and risk
  • Find out how streaming data empowers RBC to be data driven
  • Download Confluent Platform, the leading distribution of Apache Kafka^®

  
  
  https://www.confluent.io/blog/event-streaming-new-big-thing-finance
• 

  Troubleshooting KSQL – Part 1: Why Isn’t My KSQL Query Returning Data?
  ksql stream processing avro troubleshooting json ksql server • • sorangutan  

  1
  1
  Posts
  101
  Views

  

  Introduction

  KSQL is the powerful SQL streaming engine for Apache Kafka^®. Using SQL statements, you can build powerful stream processing applications. In this article, we’ll see how to troubleshoot some of the common issues that people encounter with KSQL. More advanced topics such as digging deeper into the internals for more advanced troubleshooting will be covered by a subsequent article.

  You can use this article as a reference, or you can follow along with the code examples to try them out as we go. Using Docker and Docker Compose, we can easily provision an environment in which we explore and try out the different techniques and tools. The environment includes a data generator sending continuous stream of events into a Kafka topic that we will use for testing. You can find all the necessary code on GitHub.

  So, now let’s dive in and start exploring what to do when things aren’t working…

  Why isn’t my KSQL query returning data?

  Probably the most common question in the Confluent Community Slack group’s #ksql channel is:

  Why isn’t my KSQL query returning data?

  That is, you’ve run a CREATE STREAM, but when you go to query it…

  ksql> SELECT * FROM MY_FIRST_KSQL_STREAM;

  …nothing happens. And because KSQL queries are continuous, your KSQL session appears to “hang.” That’s because KSQL is continuing to wait for any new messages to show you. So if your run a KSQL SELECT and get no results back, what could be the reasons for that?

  The answer usually comes down to one of five main reasons (the first four of which are variations on a theme):

  1. No data in the source topic
  2. No new data arriving in the topic
  3. KSQL consuming from a later offset than for which there is data
  4. Data is read from the topic, but none of it matches the predicate specified
  5. Deserialization errors in reading the data

  Let’s look at each of these in turn and how to diagnose them.

  No data in the source topic

  To start with, we believe we’ve got data in a topic called ratingz, so we register it with KSQL as a STREAM:

  ksql> CREATE STREAM RATINGS (rating_id BIGINT, \
                               user_id BIGINT, \
                               stars INT, \
                               route_id BIGINT, \
                               rating_time BIGINT, \
                               channel VARCHAR, \
                               message varchar) \
          WITH (KAFKA_TOPIC='ratingz', \
          VALUE_FORMAT='JSON');
  
   Message
  ----------------
   Stream created
  ----------------
  ksql>

  Then, we query the new stream:

  SELECT rating_id, user_id, message FROM RATINGS;

  No data comes back—time to do some detective work. To exit from the running SELECT statement, press Ctrl-C.

  First, we need to confirm which Kafka topic we’re using to drive the stream. We think we know (we just ran the CREATE STREAM, right?), but as with any good troubleshooting approach the key is methodically working through the possibilities.

  So, using DESCRIBE EXTENDED, we can verify the topic from which the stream is sourced:

  ksql> DESCRIBE EXTENDED RATINGS;
  
  Name                 : RATINGS
  Type                 : STREAM
  Key field            :
  Key format           : STRING
  Timestamp field      : Not set - using <ROWTIME>
  Value format         : JSON
  Kafka topic          : ratingz (partitions: 1, replication: 1) (1)
  [...]

  (1)The Kafka topic used by this stream

  Our source topic is ratingz. Now, we will step away from KSQL and use another Kafka consumer to verify if there is data in the topic. My preferred tool here is kafkacat, but you can use other tools, such as kafka-console-consumer, if you like. Invoking kafkacat shows:

  $ docker run --network ksql-troubleshooting_default --tty --interactive --rm \
            confluentinc/cp-kafkacat \
            kafkacat -b kafka:29092 \
            -C -t ratingz \
            -o beginning
  % Reached end of topic ratingz [0] at offset 0

  The -o beginning argument tells kafkacat to go back to the beginning of the topic, and -C to read (consume) the messages. Reached end of topic shows that there’s no data to read. No data means that KSQL isn’t going to be showing anything in the output of a SELECT.

  So in this case, there’s no data in the source topic. Turns out we mistook the topic name, and used ratingz instead of ratings! D’oh!

  No new data arriving in the topic

  Let’s fix our stream and use the ratings topic (with an s this time). Register it with KSQL, dropping the previous version first:

  ksql> DROP STREAM RATINGS;
  
   Message
  ------------------------------
   Source RATINGS was dropped.
  ------------------------------
  ksql> CREATE STREAM RATINGS (rating_id BIGINT, user_id BIGINT, stars INT, route_id BIGINT, rating_time BIGINT, channel VARCHAR, message varchar) WITH (KAFKA_TOPIC='ratings', VALUE_FORMAT='JSON');
  
   Message
  ----------------
   Stream created
  ----------------
  ksql>

  As before, try to query the stream, and you’ll find that there’s no data being returned:

  SELECT rating_id, user_id, message FROM RATINGS;
  
  [ no output returned from KSQL ]

  Like last time, first verify the source topic for the stream that we’re querying:

  ksql> DESCRIBE EXTENDED RATINGS;
  [...]
  Kafka topic          : ratings (partitions: 1, replication: 1)

  And use kafkacat to check if there’s any data in it:

  $ docker run --network ksql-troubleshooting_default --tty --interactive --rm \
            confluentinc/cp-kafkacat \
            kafkacat -b kafka:29092 \
            -C -t ratings \
            -o beginning
  {"rating_id":1,"user_id":2,"stars":1,"route_id":2350,"rating_time":1537182554356,"channel":"web","message":"thank you for the most friendly, helpful experience today at your new lounge"}
  {"rating_id":2,"user_id":10,"stars":3,"route_id":4161,"rating_time":1537182555220,"channel":"web","message":"more peanuts please"}
  [...]

  It turns out that there are thousands of messages in the topic. But, by default, KSQL reads from the end of a topic, and no new messages are being written in the topic. As soon as new messages are sent to the topic, the SELECT returns results.

  ksql> SELECT rating_id, user_id, message FROM RATINGS;
  1 | 8 | (expletive deleted)
  2 | 19 | more peanuts please
  3 | 8 | meh
  [...]

  So here we just needed to feed the topic more data. What about if you want to look at data already in the topic? That’s what we’ll look at in the next section.

  KSQL consuming from a later offset than for which there is data

  Kafka is an immutable log of events, and data is persisted according to the retention settings. When an application reads data from a Kafka topic, the data remains in place, but the offset in the log at which that particular application has read up to is recorded. Another application can read the same data from the same topic with complete independence from the first. The main thing is that there is a log of data, and consuming applications choose the point on the log at which they want to start reading.

  When KSQL reads data from a topic, it will by default read from the latest offset—that is to say, only new messages arriving in the topic after the topic is registered in KSQL will be read.

  You can verify the offset setting using LIST PROPERTIES:

  ksql> LIST PROPERTIES;
  
   Property                                               | Value
  ------------------------------------------------------------------------------------------------------------------------
  [...]
   ksql.streams.auto.offset.reset                         | latest
  [...]

  Oftentimes—and particularly in testing and development—you’ll want to read the data that already exists in a topic. To tell KSQL to do this, change the offset property:

  ksql> SET 'auto.offset.reset'='earliest';
  Successfully changed local property 'auto.offset.reset' from 'null' to 'earliest'
  ksql>

  Now when you run a SELECT, KSQL will return the data from the beginning of the topic. The SELECT will still run continuously, so if there is new data arriving, you’ll see that—and if there isn’t, the SELECT will just hang and wait for new data (or for you to cancel the query).

  If you want KSQL to read from the beginning of your topic, you must configure the offset from which it will read by runningSET 'auto.offset.reset'='earliest';.

  Data is read from the topic, but none of it matches the predicate specified

  Since KSQL is based on SQL, you can restrict the data you want to return using predicates. These include WHERE and HAVING, as well as by implication JOIN.

  If your query is not returning data and you think it should, and you’ve followed the other advice in this article already, then take a standard troubleshooting course of methodically stripping back its components until it does.

  Consider this example:

  SELECT * FROM RATINGS WHERE STARS > 5

  But if no star ratings are above 5, no data will be returned. To debug this, perhaps remove the predicate and bring in the STARS column to verify the data:

  ksql> SELECT STARS FROM RATINGS;
  
  3
  3
  2
  1

  Or, use KSQL’s aggregate functionality to look at the maximum value:

  ksql> SELECT CHANNEL, MAX(STARS) FROM RATINGS GROUP BY CHANNEL;
  android | 2
  iOS | 4
  android | 2

  From just a sample of the data, it looks like the issue here could be that STARS is never greater than 5, and thus the SELECT in the example above will never return data.

  Deserialization errors in reading the data

  Data in Kafka is just bytes. It’s up to the producer how it serializes the source message; the consumer (which is KSQL here) needs to deserialize using the same method. Common serialization formats include Avro, JSON, etc.

  If KSQL cannot deserialize message data, it will not write anything to the SELECT results. If this happens, you could have checked the three situations above and ruled them out but still not have any data returned to your SELECT.

  Here’s a simple example using one of the existing internal topics called _confluent-metrics. Let’s register it using a fictional schema that we believe to be correct for the purposes of this example, and declare the serialization format of the message values to be JSON (Tip: It’s not JSON!).

  CREATE STREAM METRICS (col1 int, col2 int, col3 varchar) \
    WITH (KAFKA_TOPIC='_confluent-metrics', VALUE_FORMAT='JSON');

  Taking the lesson from above, set the offset to earliest so that we will definitely pull all the messages, and run a SELECT:

  ksql> SET 'auto.offset.reset'='earliest';
  Successfully changed local property 'auto.offset.reset' from 'earliest' to 'earliest'
  ksql> SELECT * FROM METRICS;

  So … no results coming back. Let’s go through the checklist. We can check off the offset already, as we’ve specifically set that.

  1. What topic are we querying?

     ksql> DESCRIBE EXTENDED METRICS;
     [...]
     Kafka topic          : _confluent-metrics (partitions: 12, replication: 1)

  2. Is there any data in it?

     $ docker run --network ksql-troubleshooting_default --tty --interactive --rm \
               confluentinc/cp-kafkacat \
               kafkacat -b kafka:29092 \
               -C -t _confluent-metrics \
               -o beginning -c 1 (1)
                                                                                                                                                  ��, 
     
             kafka.logSizeLog"$partition.9.topic.__consumer_offsets*Akafka.log:type=Log,name=Size,topic=__consumer_offsets,partition=90 
     
             kafka.logSizeLog"$partition.8.topic.__consumer_offsets*Akafka.log:type=Log,name=Size,topic=__consumer_offsets,partition=80 
     
             kafka.logSizeLog"$partition.7.topic.__consumer_offsets*Akafka.log:type=Log,name=Size,topic=__consumer_offsets,partition=70 
     
             kafka.logSizeLog"$partition.6.topic.__consumer_offsets*Akafka.log:type=Log,name=Size,topic=__consumer_offsets,partition=60 
             [...]

     (1)The -c 1 argument tells kafkacat to just return the one message and then exit.

  Per the checklist, there is data, we’re querying the correct topic and have set the offset back to the beginning … but why isn’t KSQL returning data?

  Well, the data we can see from the output of kafkacat clearly isn’t JSON, which is what we declared in the CREATE STREAM command. If we go to the KSQL server log file, you’ll see a whole bunch of these deserialization errors:

   [2018-09-17 12:29:09,929] WARN task [0_10] Skipping record due to deserialization error. topic=[_confluent-metrics] partition=[10] offset=[70] (org.apache.kafka.streams.processor.internals.RecordDeserializer:86)
   org.apache.kafka.common.errors.SerializationException: KsqlJsonDeserializer failed to deserialize data for topic: _confluent-metrics
   Caused by: com.fasterxml.jackson.core.JsonParseException: Unexpected character ((CTRL-CHAR, code 127)): expected a valid value (number, String, array, object, 'true', 'false' or 'null')
    at [Source: (byte[]) ����, 
    
  [...] [truncated 1544 bytes]; line: 1, column: 2]
      at com.fasterxml.jackson.core.JsonParser._constructError(JsonParser.java:1804)
      at com.fasterxml.jackson.core.base.ParserMinimalBase._reportError(ParserMinimalBase.java:669)
      at com.fasterxml.jackson.core.base.ParserMinimalBase._reportUnexpectedChar(ParserMinimalBase.java:567)
      at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._handleUnexpectedValue(UTF8StreamJsonParser.java:2624)
      at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._nextTokenNotInObject(UTF8StreamJsonParser.java:826)
      at com.fasterxml.jackson.core.json.UTF8StreamJsonParser.nextToken(UTF8StreamJsonParser.java:723)
      at com.fasterxml.jackson.databind.ObjectMapper._readTreeAndClose(ObjectMapper.java:4042)
      at com.fasterxml.jackson.databind.ObjectMapper.readTree(ObjectMapper.java:2571)
      at io.confluent.ksql.serde.json.KsqlJsonDeserializer.getGenericRow(KsqlJsonDeserializer.java:88)
      at io.confluent.ksql.serde.json.KsqlJsonDeserializer.deserialize(KsqlJsonDeserializer.java:77)
      at io.confluent.ksql.serde.json.KsqlJsonDeserializer.deserialize(KsqlJsonDeserializer.java:45)
      at org.apache.kafka.common.serialization.ExtendedDeserializer$Wrapper.deserialize(ExtendedDeserializer.java:65)
  [...]

  You can see from the stack trace that it’s using the JSON deserializer (as you’d expect, given our VALUE_FORMAT configuration). You can also see from the sample message ([Source: (byte[]) ����, ) in the log output that the JsonParseException cites that the message clearly isn’t JSON.

  If you hit this problem, then you need to synchronize your serialization and deserialization formats. KSQL supports delimited (CSV), JSON or Avro. Protobuf users can check out KLIP-0, which proposes adding Protobuf support to KSQL.

  Not all of the messages from my topic are shown in KSQL

  Our previous example referred to cases where no messages are being returned, but you may also see cases where only some of the messages are shown, and it could be the same root cause—serialization—but with a different slant. Instead of simply getting your serialization format wrong, maybe you chose the right serialization format, but some malformed messages exist on the topic.

  In this simple example, we’ll put some data onto a new topic, using JSON but with some malformed messages:

  docker run --interactive --rm --network ksql-troubleshooting_default \
      confluentinc/cp-kafkacat \
      kafkacat -b kafka:29092 \
              -t dummy_topic \
              -P <<EOF
  {"col1":1,"col2":16000}
  {"col1":2,"col2:42000}
  {"col1":3,"col2":94000}
  EOF

  Note that the second message is invalid JSON, as it’s missing a " after the field name (col2).

  Next, register the topic in KSQL:

  ksql> CREATE STREAM DUMMY (COL1 INT, COL2 VARCHAR) \
          WITH (KAFKA_TOPIC='dummy_topic', VALUE_FORMAT='JSON');
  
   Message
  ----------------
   Stream created
  ----------------

  Set the offset to earliest so that we will definitely pull all the messages, and then run a SELECT:

  ksql> SET 'auto.offset.reset'='earliest';
  Successfully changed local property 'auto.offset.reset' from 'none' to 'earliest'
  ksql> SELECT * FROM DUMMY;
  1537186945005 | null | 1 | 16000
  1537186945005 | null | 3 | 94000

  Note that we only get two messages even though there are three on the topic.

  If you check out the KSQL server log, you’ll see:

  [2018-09-17 13:03:13,662] WARN task [0_0] Skipping record due to deserialization error. topic=[dummy_topic] partition=[0] offset=[1] (org.apache.kafka.streams.processor.internals.RecordDeserializer:86)
  org.apache.kafka.common.errors.SerializationException: KsqlJsonDeserializer failed to deserialize data for topic: dummy_topic
  Caused by: com.fasterxml.jackson.core.io.JsonEOFException: Unexpected end-of-input in field name
   at [Source: (byte[])"{"col1":2,"col2:42000}"; line: 1, column: 45]
     at com.fasterxml.jackson.core.base.ParserMinimalBase._reportInvalidEOF(ParserMinimalBase.java:594)
     at com.fasterxml.jackson.core.json.UTF8StreamJsonParser.parseEscapedName(UTF8StreamJsonParser.java:1956)
     at com.fasterxml.jackson.core.json.UTF8StreamJsonParser.slowParseName(UTF8StreamJsonParser.java:1861)
     at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._parseName(UTF8StreamJsonParser.java:1645)
     at com.fasterxml.jackson.core.json.UTF8StreamJsonParser.nextFieldName(UTF8StreamJsonParser.java:999)
     at com.fasterxml.jackson.databind.deser.std.BaseNodeDeserializer.deserializeObject(JsonNodeDeserializer.java:247)
     at com.fasterxml.jackson.databind.deser.std.JsonNodeDeserializer.deserialize(JsonNodeDeserializer.java:68)
     at com.fasterxml.jackson.databind.deser.std.JsonNodeDeserializer.deserialize(JsonNodeDeserializer.java:15)

  Note the partition and offset shown in the error message (partition=[0] offset=[1]).

  Head back to the ever-versatile kafkacat and run the following:

  docker run --network ksql-troubleshooting_default \
            --tty --interactive --rm \
            confluentinc/cp-kafkacat \
            kafkacat -b kafka:29092 -C -K: \
            -f '\nKey: %k\t\nValue: %s\n\Partition: %p\tOffset: %o\n--\n' \ (1)
            -t dummy_topic \
            -o 1 \ (2)
            -p 0 \ (3)
            -c 1   (4)

  (1)-f to format the output and show some nice metadata
  (2)-o 1 start at offset 1
  (3)-p 0 read from partition 0
  (4)-c 1 consume just one message

  The output is:

  Key:
  Value: {"col1":2,"col2:42000}
  Partition: 0    Offset: 1
  --

  This shows us, if we were in any doubt, that the message value is not valid JSON and thus can’t be consumed by KSQL.

  If KSQL is not showing all of your messages, then check that they are in a valid format for the declared serialization. So if you’ve set VALUE_FORMAT='JSON', then check that each message is indeed valid JSON.

  Locating KSQL server logs

  KSQL writes most of its logs to stdout by default. If you’re running KSQL using Docker, then you’ll find the output in the container logs themselves. For example:

  • docker logs 483b1958efc4
  • docker-compose logs ksql-server

  Using the Confluent CLI, you can run the following:

  • confluent log ksql-server

  If you’ve installed Confluent Platform using RPM/DEB, then you should find the logs under /var/log/confluent/.

  Need more help?

  Still stuck and need more help? Here are several places to turn:

  • Get support for KSQL as part of Confluent Enterprise
  • Seek community support for KSQL:
    • Ask questions in the Confluent Community Slack #ksql channel
    • Search for similar issues on GitHub
    • Raise a new issue on GitHub if it doesn’t already exist

  Other articles in this series:

  • Coming soon: “Troubleshooting KSQL – Part 2: What’s happening under the covers?” dives into KSQL internals, and looks at tools we can use for monitoring and examining the behavior of running queries and KSQL itself.

  
  
  https://www.confluent.io/blog/troubleshooting-ksql-part-1
• 

  Real-Time Presence Detection at Scale with Apache Kafka on AWS
  aws apache kafka kafka streams use case use cases real time • • sorangutan  

  1
  1
  Posts
  108
  Views

  

  This guest post is written by Eugen Feller, a staff software engineer and technical lead at Zenreach, where he is responsible for the data infrastructure. He holds a Ph.D. with honors in computer science from the French Institute for Research in Computer Science and Automation (INRIA).

  Zenreach provides a platform that leverages Wi-Fi and enables merchants to better understand and engage with their customers. Most of our product features depend on one basic building block: real-time customer presence detection.

  In this blog post, we share our journey of implementing a real-time presence detection platform and how we leveraged Kafka Streams. Being a startup, our platform started off as a set of Python scripts, moved to Spark Streaming and ultimately ended up being re-implemented using Kafka Streams in early 2017 for ease of operations and cost reasons. Since switching to Kafka Streams (we started with version 0.10.2.1 and are currently on 0.11.0.1), we have learned a lot about the Apache Kafka^® ecosystem. We have implemented logging, monitoring and state management, and have gained significant understanding of which parameters need to be tuned. We have rolled out a robust platform that is highly available and fault-tolerant and offers a degree of observability that we never had before. This, along with the fact that all of our streaming services are now running in containers on Amazon Elastic Container Service (AWS ECS), has helped us save thousands of dollars per month. Finally, we have noticed a significant increase in developer productivity.

  Our use case

  As we receive Wi-Fi device data from our routers, we need to perform presence detection. When we talk about presence detection, we are primarily interested in detecting walk-ins. A walk-in indicates that a customer is actually at the store.

  Once we have detected presence, we know that a specific device was sighted at the business. However, oftentimes this is not enough. We need to enrich this walk-in with customer contact information (e.g., email) to make it useful. Contact information is emitted to Kafka by our customer relationship management (CRM) service each time a customer logs in with our Wi-Fi portal.

  Finally, we need to maintain a walk-in count for each business and contact, keeping it updated as customers visit our businesses.

  Why Kafka Streams

  In our early days, we quickly implemented the aforementioned use case with a number of Python scripts running on a single machine. However, as the company quickly scaled by adding more businesses, the Python scripts were no longer sufficient to support the increasing probe throughput. As a result, we re-implemented our system using a monolithic Spark Streaming application. This Spark Streaming application was detecting walk-ins, enriching walk-ins with contact information, maintaining walk-in counts, storing walk-in counts in our database and emitting presence events (e.g., walk-ins and enriched walk-ins) to Kafka. Over time, it became obvious that this approach of a gigantic Spark Streaming application had some serious drawbacks:

  • It was very hard to fix and deploy bug fixes in various parts of this application. For example, a bug in the walk-in algorithm required a redeployment of the entire application even though other parts of the application were working fine.
  • It was very hard to monitor this application, as we had no proper centralized resource monitoring and logging (stdout/stderr) on our Amazon Elastic MapReduce (AWS EMR) infrastructure.
    • We were running blind in regards to resource consumption (e.g., CPU, memory).
    • Developers had a hard time finding error logs. Indeed, they had to browse various Hadoop UIs to get to Spark executor logs. This was time and resource intensive.
    • We had no central scheduler to run the Spark application. Consequently, our application was started as a cron job on a dedicated Amazon EC2 machine and restarted by the same cron job if it failed. Finding out why the application crashed was cumbersome.
  • The EMR infrastructure was not plugged into our CI/CD system; hence, rolling out new features and bug fixes to production was manual and error prone.
  • The cost of running entire EMR infrastructure for just one use case was huge. A single EMR cluster was costing us thousands of dollars a month.

  It was pretty clear that we could no longer continue operating with these drawbacks, so we decided to take the following actions:

  • Split the monolithic Spark application into multiple applications, according to their logical responsibility, to make bug fixing easier
  • Plug the applications into our centralized Prometheus-based monitoring and alerting system to gain resource usage observability
  • Integrate all applications with Rollbar and Logentries, the two logging solutions that we use for our core services
  • Integrate all applications into our CI/CD system in order to release them along with our core services
  • Significantly cut down on our EMR costs by running all applications in containers along our core services

  Given these changes and the fact that we already had Kafka running on our infrastructure, we looked for a streaming solution that would integrate smoothly. We wanted to do streaming without additional infrastructure, allowing us to run the streaming services with our core services. We had learned it the hard way that maintaining additional infrastructure (e.g., EMR) with a limited amount of engineering resources is nearly impossible, especially for a company of our size (100-200 people). We liked the idea that Kafka Streams is a client library built on top of existing Kafka features (e.g., consumer groups). Despite being relatively new at the time, Kafka Streams provided the features we needed (e.g., state stores, joins, windowing and punctuation).

  Our system

  Our system architecture is shown below:As customers visit businesses, routers collect device (e.g., phone) probe requests and send them to our location service. Each probe request includes a unique device identifier used to detect presence. Our presence platform is made up of four services: Walk-In, DeviceToContact, Enriched Walk-In and Walk-In Count.

  The Walk-In service detects devices and emits walk-in events to Kafka.

  The DeviceToContact service reads CRM events, such as a portal login, and turns them into device-to-contact mappings. A device-to-contact mapping identifies the businesses at which the device was seen and the contact information that was used by the device-to-portal login. Device-to-contact mapping events are emitted to Kafka.

  The Enriched Walk-In service takes the walk-ins and enriches them with contact information by joining them with the device-to-contact mapping events. The enriched walk-ins are emitted back to Kafka.

  The Walk-In Count service maintains a walk-in counter for each business-contact pair. It reads enriched walk-ins and updates the internal walk-in counter for each business-contact pair. For each update, it emits walk-in count events to Kafka with the most recent counts.

  We have downstream systems that consume walk-ins, enriched walk-ins and walk-in count topics for various product features. Examples include sending marketing emails to returning customers or showing customers that are currently in the store on our mobile application.

  Lessons learned

  Scaling your services

  One important consideration that led us to choose Kafka Streams was ease of scaling. Kafka Streams-based services can scale horizontally merely by increasing the number of service instances. In this context, it is important to understand the Kafka Streams threading model which is made up of two concepts: stream threads and stream tasks. Stream threads are real Java threads. They can be seen as pools of available resources for data processing. Stream tasks are abstract units of computation that are managed by stream threads. Each stream task processes a maximum of one partition per topic. A stream thread can process multiple stream tasks.

  Internally, a stream thread initializes a Kafka consumer that fetches records for multiple topic partitions. Records of multiple stream tasks/partitions are processed by a stream thread sequentially by passing them through the defined topology. It is important to make sure that stream tasks are well balanced across the stream thread. Otherwise, throughput will be negatively impacted. A single stream task that is stuck in a blocking call to an external service can also stall the entire stream thread, and long blocking calls should be avoided.

  Kafka Streams provides the parameter num.stream.threads that can be used to control the number of stream threads per service instance. We had to adjust this parameter to make sure that we achieve best processing throughput on our services. For example, if a service is deployed with four instances and num.stream.threads per instance is set to four, in total 16 stream threads are available for stream processing. If a topic has 32 partitions, Kafka Streams will create 32 stream tasks for 32 partitions, and assign two stream tasks per stream thread.

  Persistent storage

  Most of our streaming services are stateful. For example, they need to keep track of walk-ins and walk-in counts. Kafka Streams uses changelog topics and RocksDB-backed state stores to enable stateful services. When we first started with our platform, we deployed streaming services on ECS the same way we deployed stateless services. This caused a major issue during releases.

  For a release, we typically do a blue-green deploy of our services to minimize downtime. Therefore, services get restarted and can be scheduled on any machine in our ECS cluster that has sufficient resources in terms of CPU and memory. This results in a RocksDB state loss, as the newly selected machine no longer has the previous state store content. In this case, Kafka Streams starts to recreate the state store content from the backing changelog topic. With the amount of state that our streaming services are managing, it took hours for the streaming services to become stable after the release was complete.

  To resolve this issue, we attached EBS GP2 volumes to a subset of our ECS machines and used ECS placement constraints to pin our stateful streaming services to the machines. We then used Docker volumes and the Kafka Streams state.dir configuration parameter to ensure that our Kafka Streams services make use of the persistent storage. This has enabled us to make releases of Kafka Streams services instant with no downtime.

  Fault tolerance

  In the event of a service instance start, stream tasks need to be assigned. Stream tasks are sticky to their state. On startup, Kafka Streams checks the local state directly state.dir and if it finds the existing state, it tries to reuse it. Specifically, it assigns stream tasks to instances that have state available. This turned out to be extremely useful for us as we use EBS volumes for persistent storage. If stream tasks were not sticky, they would need to replay state from the corresponding changelog topics, thus defeating the purpose of persistent storage.

  In the event of a service instance crash, stream tasks need to be reassigned to different service instances. This causes a temporary drop in throughput as the resulting consumer group rebalancing may take some time. Indeed, if state stores are involved, state needs to be replayed from the changelog topics. In order to reduce the recovery time in the event of a service instance crash, Kafka Streams provides the parameter num.standby.replicas, which defaults to zero. This parameter controls the number of standby stream tasks that are available to take over if primary task failure occurs.

  Standby stream tasks do not actively process records. Instead, they just keep replicating the state of primary stream tasks. Kafka Streams will make sure that standby stream tasks are placed on distinct service instances from where the primary stream tasks are hosted. We have adjusted this parameter to one in production. As a result of this, each stream task has a standby stream task available to take over in the event of a service instance crash. This has helped us to greatly reduce rebalancing duration and downtime during outages.

  As another consideration, it may be desirable to further place stream tasks across multiple availability zones (AZs). AZs are typically located in different physical locations. This would further improve fault tolerance in cases where one AZ goes down, as no two replica stream tasks would be residing on the same AZ. As of today, Kafka Streams does not provide such a feature. However, it is planned and currently in development (see KAFKA-6718).

  Rebalancing

  Rebalancing happens when Kafka assigns partitions to consumers. This is a normal process as part of Kafka consumer group protocol and should ideally happen as rarely as possible, such as during a service restart or service instance failure. When rebalancing happens too frequently, it can result in some serious throughput degradation. Rebalancing can happen for many reasons and unfortunately we have encountered quite a few.

  Failure to send heartbeats to coordinator on time

  If your service does not manage to send heartbeats on time to the current group coordinator—one of the Kafka brokers—Kafka would consider the consumer as dead and trigger a rebalancing process. Fortunately, Kafka Streams and more specifically the Kafka consumer has two parameters to control the heartbeat interval: heartbeat.interval.ms and session.timeout.ms. The first parameter controls how often heartbeats should be sent to the coordinator. The second parameter controls how long the coordinator should wait until a consumer is considered dead, in which no heartbeat is received. In order to increase session.timeout.ms on the clients, we ended up increasing the group.max.sesion.timeout.ms parameter on our Kafka brokers. This parameter with a default of 30 seconds controls the maximum session timeout that can be used by the clients.

  Long garbage collection pauses

  Many of our services are memory intensive with extensive usage of state stores and in-memory caches. If your service takes more than 10 seconds in garbage collection (GC) pauses, it can trigger the consumer to be kicked out of the consumer group and thus force a rebalancing process among the remaining consumers. We have reacted to such situations by giving more memory to our streaming jobs and adjusting the heap sizes. We have also made sure that we close all state store iterators. Otherwise, the memory consumption would keep increasing. Our services are using Java 8. We have switched to the Garbage-First (G1) GC to further reduce lengthy pause times. Finally, to help dealing with longer GC pauses, we have decreased the heartbeat interval and increased the session timeout.

  Long blocking calls inside the Kafka Streams API

  Some of our downstream services are reading data from Kafka like walk-in counts and writing data back to a database to be shown by the UI. In the early days, we simply wrote such data synchronously into the database while reading Kafka, such as inside a foreach call. As it turns out, each stream processor thread in Kafka Streams calls poll() on Kafka to fetch records of potentially multiple partitions. Internally, the stream processor thread logic is sequential. In other words, a stream processor thread processes records of potentially many partitions one at a time. Depending on the amount of data to be processed and latency in writing to the database, it can take a while to write out all of the data to the database.

  We had run into several issues where processing all the records took too long and consumers were kicked out of the consumer group, which triggered rebalancing. We found that a Kafka consumer has a parameter called max.poll.interval.ms that can be used to control the maximum delay between consecutive poll() invocations. We also learned that the parameter max.poll.records is extremely useful as it allow us to control how many records will be fetched from Kafka in each poll() request. To control the amount of time that clients will wait for Kafka to respond with data, we used the request.timeout.ms parameter.

  As our Kafka cluster became more loaded, some fetch requests were timing out. By tuning these parameters and making all our database calls asynchronous, we were able to greatly improve the service stability. Note that max.poll.interval.ms is set to MAX_VALUE by default. In the beginning, we made the mistake of changing this default to a lower value which caused more harm than good. We advise to keep it as is.

  Initial consumer group rebalancing

  As we were deploying and releasing our services in containers, we noticed that a lot of rebalancing was happening during the release periods. This was due to the fact that in a container environment, service instances typically come online in a non-deterministic order. Moreover, the time between the service instances becoming available depends on the resource availability in our ECS cluster. Each time a service becomes available, it triggers a consumer group rebalancing in order to get partitions assigned.

  Our services are deployed with many instances for performance and fault tolerance reasons. We noticed that a significant amount of rebalancing had to happen in order for the service as a whole to become stable. Fortunately, Kafka broker offers the configuration parameter group.initial.rebalance.delay.ms to control the rebalancing delay. This allowed us to delay the initial rebalancing until the majority of our service instances became available. We increased the default value of three seconds for this parameter to 60 seconds on all our Kafka brokers, which has helped to significantly reduce the number of rebalance rounds.

  Logging, monitoring and alerting

  One of the big reasons why we moved to Kafka Streams is ease of logging with stdout and stderr. Kafka Streams services are treated the same way as our core services, so we can leverage the same logging infrastructure. We initially started out by routing all logs to Amazon CloudWatch. However, this turned out to be cumbersome for developers as CloudWatch offers limited search capabilities. Furthermore, we wanted to route all logs with a logging level greater than WARN to Slack so that we could react to them quickly. Over time, we added support to write all logs to a cloud service called Logentries. This allowed us to easily search and analyze logs. For routing critical exceptions to Slack, we used another service called Rollbar.

  In addition, monitoring is a critical piece of running a production infrastructure. We need to know the following and more:

  • How much memory our services are consuming
  • How much time services spent in GC
  • Current throughput on all services
  • How far the services are lagging behind
  • How much storage headroom we have left on our EBS volumes
  • How much storage is left on the Kafka brokers
  • RocksDB performance

  We use Prometheus across the board to collect, visualize and alert on critical metrics from all our services. We have also instrumented our services to report custom metrics. For example, we have used the KafkaStreams#setStateListener API to get notified about state transitions (e.g., REBALANCING to RUNNING, and vice versa).

  To gain insights into various JVM-related metrics like GC time, memory consumption, RocksDB performance and throughput per Kafka topic, we found the Prometheus JMX exporter to be very useful. We run it alongside all our streaming services and Kafka itself. For reporting lag on various Kafka topics we have deployed the Prometheus Kafka Consumer Group Exporter. To know how much disk storage is left on our machines, we use the Prometheus Node Exporter, which is deployed on all our ECS instances. For container-level CPU, memory and network utilization metrics, we run cAdvisor. All of this combined provides us significant insights into the current operations of our presence detection platform.

  

  Figure 1. Rebalancing rate graph for one of the services

  

  Figure 2. EBS disk usage percentage graph for one of the services

  Data backfilling

  One of the important requirements in our streaming pipeline is data backfilling. We present two examples where data backfills are important in our pipeline: (1) walk-in count and (2) device to contact.

  Walk-in count example

  When a topic is deleted or corrupted, it is required to provide a base walk-in count to the walk-in count service. Base walk-in counts are the state from which we start incrementing walk-in counts. Walk-in counts are stored in a state store. Initially, we used a Cassandra table which was backfilled from our Amazon S3 archives with the latest walk-in counts when needed. The walk-in count service consulted this table in the event that a walk-in count is unknown. However, this required an ETL job that reads the walk-in counts from our primary database MongoDB and dumps them to Cassandra. This did not scale well with the amount of walk-in counts that we had and resulted in all sorts of issues while writing to Cassandra to the point where Cassandra would go down during ETL.

  After learning the Kafka internals, we decided to take another approach. It was possible to backfill data into the changelog topic that backs the walk-in counts state store directly. Kafka Streams would automatically bootstrap the state store to the end of the topic. This turned out to be very useful for us. We wrote another ETL job that simply reads walk-in count backups from our S3 archives and backfills them directly to the changelog topic.

  DeviceToContact Example

  Our DeviceToContact service maintains a mapping from device identifier to contacts. This mapping is important as it is used by the Walk-In service to enrich walk-ins with contact information. Internally, the DeviceToContact service reads CRM event streams as KTables and performs a join. The resulting device-to-contact mappings are stored in a log-compacted topic.

  Similarly to the previous example, a backfill may be required in case the device-to-contact mappings were computed wrongly and/or data was lost. This can be achieved by an ETL job that regenerates CRM events from our database and writes the data back to dedicated backfill topics. The backfill topics can be consumed by our service in addition to regular topics. KTables unlike KStreams do not support consuming from multiple topics, as they would not be able to guarantee record order per key. We ended up creating KStreams out of our input topics and the corresponding backfill topics that were turned into KTables (see here for more details).

  Data corruption

  Each record in a Kafka topic has a cyclic redundancy check (CRC) checksum associated with it. Data corruption may occur if the records content was modified such that the stored CRC checksum no longer matches the recomputed one. This results in a client-side CorruptRecordException thrown in the Kafka consumer fetch layer as it attempts to validate consumed records (see KAFKA-6977). We have run into CRC checksum failures primary with log-compacted topics, leading services to crash and be unable to process records. A Kafka consumer does not have a way to skip corrupted records (see KAFKA-7405).

  Using the KafkaStreams#setUncaughtExceptionHandler API was not an option for us, as the exception handler is called after the stream thread has already died. Similarly, using the default.deserialization.exception.handler setting (see KIP-161) is not an option, as it is called after data has been fetched. Hence, it is not adequate to handle CRC checksum failures that happen in the data fetch layer.

  Although we were unable to reproduce the CRC checksum failures, we suspect they might be related to a bug in the log compaction mechanism and/or use of SSD disks on our Kafka brokers using version 0.10.2.1 at the time. Fortunately, we found that Kafka provides great consumer group tooling that enables resetting consumer group offsets and/or skipping bad offsets. Specifically, the bin/kafka-consumer-groups CLI can be used to manipulate consumer group offsets.

  Resetting consumer group offsets on all partitions to earliest and/or latest was not an option for us, as it would result in data loss. However, the same tool allowed us to shift offsets by a delta and/or time. We used shift offsets to skip corrupted records and ended up deploying a new Kafka 1.1.0 cluster. So far, we have not run into further CRC checksum failures. It would be great if a Kafka consumer could be extended with the ability to gracefully intercept and ignore corrupted records.

  Conclusion

  Our presence platform has evolved from a set of Python scripts to a robust streaming pipeline based on Kafka Streams. While it was a bumpy ride in the beginning, looking back at our evolution, Kafka Streams turned out to be a great choice for our use case. It has allowed us to simplify our streaming infrastructure greatly. Our streaming services can now be deployed in the same manner as our core services in containers on AWS ECS. This is a huge win, as a separate data infrastructure is no longer required. Indeed, this has allowed us to decommission our entire EMR deployment, which was costing us thousands of dollars per month.

  Deploying on ECS also greatly simplified our streaming services release process, improved developer productivity and streamlined monitoring and alerting. Our presence platform services are now released in the same manner as core services, debugged in the same manner as core services and monitored the same way as our core services via a global Prometheus monitoring and alerting system. We are very much looking forward to reworking other parts of our product to be event driven using Kafka technology as we move to the latest Kafka Streams version.

  About Apache Kafka’s Streams API

  If you have enjoyed this article, you might want to continue with the following resources to learn more about Apache Kafka’s Streams API:

  • Get started with the Kafka Streams API to build your own real-time applications and microservices
  • Walk through our Confluent tutorial for the Kafka Streams API with Docker and play with our Confluent demo applications

  We would not have been able to get where we are today without the wonderful Kafka community. I would especially like to thank Matthias J. Sax from Confluent for his amazing help in answering and clarifying many of our internal Kafka Streams questions.

  
  
  https://www.confluent.io/blog/real-time-presence-detection-apache-kafka-aws
• 

  The State of Streams – A European Adventure
  event company streaming • • sorangutan  

  1
  1
  Posts
  71
  Views

  

  I have good news to share! For the first time, this October Confluent will bring to Europe two events dedicated to streaming technology: Confluent Streaming Event in Munich and Confluent Streaming Event in Paris. We will share about use cases from a variety of industries, and teach developers, architects and business stakeholders how to build a faster, more responsive digital business.

  These events are for developers taking their first steps in the streaming world, experts who want to learn about the latest features added to Confluent Platform and architects who want to take the streaming platform to the next level.

  I am honored and excited to deliver the opening keynotes in both cities. I can’t wait to meet the local Apache Kafka^® communities and exchange thoughts about Kafka and the evolution its use cases.

  I’ll talk about streaming platforms, why they are changing the world and how they evolved. I’ll share real-world examples ranging from early use cases that companies first adopted to the more exciting, cutting-edge work we currently see. I’ll also cover examples of real-time analytics, event-driven microservices, IoT and fraud detection. Hopefully these stories will inspire you on your adoption journey too. And finally, I’ll share my take on future industry trends and how our platform will evolve to meet new challenges.

  I also want to highlight some other great speakers on the agenda for both cities who will discuss how they use Apache Kafka and Confluent Platform to conquer digital Darwinism (Dominique, I quote you here).

  Confluent Streaming Event in Munich, Oct. 9:

  • Alexander Alten-Lorenz, Chief Architect Global Platform/Technology COO-C Digital, E.ON SE
  • Christian Nicoll, Director of Platform Engineering and Operations, Generali Personenversicherungen AG – Switzerland
  • Dominique Ronde, Big Data Pilot, Allianz Technology
  • Michael Noll, Product Manager, Kafka Streams and KSQL, Confluent
  • Hans Jespersen, Vice President, WW Systems Engineering, Confluent

  Confluent Streaming Event in Paris, Oct. 11:

  • Yves Chophel, Technical Architect, Project Management Institute
  • Mickaël Saunier and Yutao Sun, Direction STET MOE Développement, STET
  • Florent Ramière, Sales Engineer, Confluent
  • Hans Jespersen, Vice President, WW Systems Engineering, Confluent

  Join us

  I invite you to join us and learn the best practices of simplifying and scaling software architectures with the streaming platform. It’ll be a great opportunity to gather together with other business enthusiasts, Kafka pioneers, Confluent users and experts to explore the unique streaming projects active across various industries today. I hope to meet you there!

  • RSVP to Confluent Streaming Event in Munich on Oct. 9, 2018
  • RSVP to Confluent Streaming Event in Paris on Oct. 11, 2018

  
  
  https://www.confluent.io/blog/state-streams-european-adventure
• 

  The Changing Face of ETL
  ksql stream processing apache kafka architecture etl • • sorangutan  

  1
  1
  Posts
  88
  Views

  

  The way in which we handle data and build applications is changing. Technology and development practices have evolved to a point where building systems in isolated silos is somewhat impractical, if not anachronistic. As the cliché of a cliché goes, data is the new oil, the lifeblood of an organization—however you want to analogize it, the successful handling and exploitation of data in a company is critical to its success.

  In this article we’re going to see a practical example of a powerful design pattern based around event-driven architectures and a streaming platform. We’ll discuss why companies are identifying real-time requirements and adopting this instead of legacy approaches to integrating systems.

  Part of this shift comes from the blurring of lines between analytic systems and transactional systems. In the old days, we built transactional systems on which our business ran, and then a separate team would be responsible for extracting the data and building analytics (or BI, or MIS or DSS, depending on your age) on that. It was done in an isolated manner and was typically a one-way feed. Analytics teams took the data, on which processing was done, to cleanse and enrich it before building reports.

  Now, we have the beginnings of an acceptance that data is not only generated by systems but when combined with other data and insights can actually be used to power systems. It’s about enabling the integration of data across departments, teams and even companies, making it available to any application that wants to do so and building this in such a way that is flexible, scalable and maintainable in the long term. And with today’s technology, we can do so in real time, at scale.

  For example, retail banks are faced with the challenge of providing a compelling customer experience to a new generation of users, whose expectations are shaped by AI-powered mobile applications from Google and Apple. In order to meet this challenge, they need to combine everything they know about their customers with state-of-the-art predictive models, and provide the results to mobile apps, web apps and physical branches. ING Bank and Royal Bank of Canada use event-driven architectures to power modern systems that do exactly that. Let’s dive into the technologies and design patterns that enable them to transform their user experience and development process at the same time.

  ETL but not as you know it

  Consider a simple example from the world of e-commerce: On a website, user reviews are tracked through a series of events. Information about these users such as their name, contact details and loyalty club status is held on a database elsewhere. There are at least three uses for this review data:

  • Customer operations: If a user with high loyalty club status leaves a poor review, we want to do something about it straight away to reduce the risk of them churning. We want an application that will notify us as soon as a review meeting this condition is met. By doing so immediately, we can offer customer service that is far superior than if we had waited for a batch process to run and flag the user for contact at a later date.
  • Operational dashboards showing a live feed of reviews, rolling aggregates for counts, median score and so on—broken down by user region, etc.
  • Ad hoc analytics on review data combined with other data (whether in a data lake, data warehouse, etc.): This could extend to broader data science practices and machine learning use.

  All of these use cases require access to the review information along with details of the user.

  One option is to store the reviews in a database, against which we then join a user table. Perhaps we drive all the requirements against the data held in this database. There are several challenges to this approach, however. Because we are coupling together three separate applications to the same schema and access patterns, it becomes more difficult to make changes to one without impacting the others. In addition, one application may start to impact the performance of another—consider a large-scale analytics workload and how this could affect the operational behavior of the customer ops alerting application.

  

  In general, we would not expect to build the three applications against the same data store. Thus, we extract the review and user data to target systems best suited for serving the dependent application. Perhaps we use a NoSQL store to drive the customer operations application, the Elastic Stack for the ops dashboard and something like HDFS, Amazon S3, Google BigQuery or Snowflake for the analytics platform.

  

   

  But now, the following questions emerge:

  • How do we populate each target from the review data in a consistent and efficient way?
  • How do we enrich the review data with user information? Do we try and do it once before loading each target, or do we repeat the process as part of loading each target? Can we do this with up-to-date versions of the customer data while also maintaining a low-latency feed?

  We also have longer-term architectural considerations. If we add new sources for the review data (e.g., mobile application, public API), how is that fed to the target systems? What if one of the target systems is offline or running slowly? How do we buffer the data or apply backpressure? If we want to add new targets for the data in the future, how easy will it be to do this?

   

  You say tomato; I say tomato

  In the past we used ETL techniques purely within the data-warehousing and analytic space. But, if one considers why and what ETL is doing, it is actually a lot more applicable as a broader concept.

  • Extract: Data is available from a source system
  • Transform: We want to filter, cleanse or otherwise enrich this source data
  • Load: Make the data available to another application

  There are two key concepts here:

  • Data is created by an application, and we want it to be available to other applications
  • We often want to process the data (for example, cleanse and apply business logic to it) before it is used

  Thinking about many applications being built nowadays, particularly in the microservices and event-driven space, we recognize that what they do is take data from one or more systems, manipulate it and then pass it on to another application or system. For example, a fraud detection service will take data from merchant transactions, apply a fraud detection model and write the results to a store such as Elasticsearch for review by an expert. Can you spot the similarity to the above outline? Is this a microservice or ETL process?

  ETL is a design pattern, applicable to a huge variety of purposes beyond traditional data warehousing from which it originated.

  
  The streaming platform in action

  Let’s see what the above e-commerce example looks like in practice when implemented using a more modern design pattern, based on a streaming platform. We’re going to use the open source Apache Kafka^® and KSQL projects to do this. KSQL is the streaming SQL engine for Apache Kafka, implemented on top of the Kafka Streams API, which is part of Apache Kafka itself.

  In our example, the events are the reviews that the users submit on the site, and these are streamed directly into Kafka. From here, they can be joined to the user information in real time with the resulting enriched data written back to Kafka. With this transformation done, the data can be used to drive the dependant applications and targets. The transformation logic is only required once, and the data is extracted from the source system only once. The transformed data can be used multiple times by independent applications. New sources and targets can be added without any change to the existing components. All of this is very low latency.

  At a high level, the design looks like this:

  • Web application emits review directly to Kafka
  • Kafka Connect streams snapshot of user data from database into Kafka and keeps it directly in sync with change data capture
  • Stream processing adds user data to the review event and writes it back to a new Kafka topic
  • Stream processing filters the enriched Kafka topic for poor reviews from VIP users and writes to a new Kafka topic
  • Event-driven application listens to a Kafka topic and pushes notifications as soon as a VIP user leaves a poor review
  • Kafka Connect streams the data to Elasticsearch for an operational dashboard
  • Kafka Connect streams the data to S3 for long-term ad hoc analytics and use alongside other datasets

  

  The benefits include:

  • Data enrichment occurs once and the enriched data is available for any consuming application
  • Processing is low latency
  • Notifications to the customer ops team happen as soon as the VIP customer leaves a poor review, leading to a much better customer experience and higher chance of retaining their business
  • Ability to scale easily by adding new nodes as required for greater throughput 

  Transform once, use many

  Often the data used by one system will also be required by another, and the same goes for data that has been through enrichment or transformation. The work that we do to cleanse the inbound stream of customer details, such as standardizing the country name, state/county identifiers and phone number formatting—all of this is useful to both the analytics platform downstream as well as any other application that deals with customer data. We can formalize this into a three-tier data model applicable to many data processing architectures. Tier 1 is the raw data. Tier 2 is the integration layer, where data is clean and standardized, available for anyone to analyze. Tier 3 is the application layer, in which data is joined, aggregated and formatted in a way that maximizes its usefulness for the needs of a specific application.

  A great pattern to adopt is to stream raw data (Tier 1) as it is transformed (Tier 2) back into Kafka. This makes that data available in real time to all applications directly. The alternative is the legacy pattern of writing transformed data down to a target (often a data lake) and having other applications pull the data from there—with the associated latency and complication to our systems architecture. Where applications come to consume and transform the data for their own purpose (Tier 3), this can also be done in Kafka. Depending on technical requirements, using other technologies such as NoSQL, RDBMS or object storage may be required.

  By streaming the transformed data back into Kafka, we receive some great advantages:

  • Separation of responsibilities between the transformation and application/system consuming that data: The latency remains low, as the transformed data that is streamed to Kafka can be streamed straight to the desired target. Even for a transformation in which you think only your application will want the transformed data, this pattern is a useful one.
  • Transformed data can be used to drive other applications: Because Kafka persists data, the same data can be used by multiple consumers—and completely independently. Unlike traditional message queues, data is not removed from Kafka once it has been consumed.
  • Single instance of the transformation code: This instance includes any associated business logic that it implements. That means a single place in which to maintain it, a standard definition for any measures derived and consistency in the data across systems. Contrast this with multiple systems each performing the same transformation logic. For the best will in the world, the code will diverge, and you will end up hunting for that needle in a haystack of why your data doesn’t reconcile between systems.

  The goal is to avoid creating 1:1 pipelines and instead create a hub with the platform at the center. Traditionally, ETL would be done on a point-to-point basis, taking data from a source system and loading it to a target one. If the data was needed elsewhere it would either be extracted twice or taken from the original target. Both of these are undesirable. The former increases the load on the source system, and the latter introduces an unnecessary dependency and coupling in the design. In short, this is how the “big-ball-of-mud” or “spaghetti” architectures start.

  

  By adopting a streaming platform, we decouple the sources and targets for data, and thus introduce greater flexibility to build upon and evolve an architecture.

  

  Let’s build it!

  So far, we’ve just looked at the theory behind a streaming platform and why one should consider it as a core piece of architecture in a data system. To learn more, you can read my InfoQ article Democratizing Stream Processing with Apache Kafka and KSQL, which builds on the concepts in this blog post and provides a tutorial for building out the e-commerce example described above. You may like to also check out the following resources:

  • Watch the online talk ETL is Dead; Long Live Streams
  • Get an introduction to KSQL
  • Download Confluent Platform and follow the quick start to begin using KSQL
  • Check out the KSQL video tutorials and KSQL hands-on tutorials
  • Ask questions in the #ksql channel of our community Slack group

  
  
  https://www.confluent.io/blog/changing-face-etl
• 

  Hands on: Building a Streaming Application with KSQL
  ksql stream processing real time • • sorangutan  

  1
  1
  Posts
  124
  Views

  

  In this blog post, we show you how to build a demo streaming application with KSQL, the streaming SQL engine for Apache Kafka^®. This application continuously computes, in real time, top music charts based on a stream of song play events.

  Does this demo sound familiar? It should. Last year Confluent produced a similar demo application using the Kafka Streams API. Since then, Confluent has released production-ready KSQL. Users can choose between Kafka Streams or KSQL depending on the use case. Both enable developers to build real-time, scalable, fault-tolerant applications with exactly-once capabilities, for Confluent Cloud and on-premises Kafka deployments. Here we take the same music application concept and rewrite it using KSQL so you can compare and contrast the experience.

   

  This tutorial accompanies:

  • KSQL Streaming Application for Music Charts screencast
  • Automated KSQL demo 

  Setup

  Note: This tutorial assumes you are running the KSQL music demo application in Docker. You may alternatively run this demo on a Confluent Platform local install just by running the start script.

  Clone this quickstart-demos repo.

  $ git clone https://github.com/confluentinc/quickstart-demos

  In Docker’s advanced settings, increase the memory dedicated to Docker to at least 8 GB (default is 2 GB).

  From the quickstart-demos/music directory, start the demo by running a single command that brings up the Docker containers. This will take less than two minutes to complete.

  $ cd quickstart-demos/music
  $ docker-compose up -d

  Do not proceed until you see the below output in the logs of Confluent Control Center.

  $ docker-compose logs -f control-center | grep -i "Started NetworkTrafficServerConnector"
  control-center                | [2018-09-06 15:03:22,518] INFO StartedNetworkTrafficServerConnector@4d48bd85{HTTP/1.1,[http/1.1]}{0.0.0.0:9021} (org.eclipse.jetty.server.AbstractConnector)

  Inspect the source Kafka data

  The Docker compose file includes a container that auto-generates source data to two topics in Apache Avro format:

  • play-events: stream of play events (“song X was played”)
  • song-feed: stream of song metadata (“song X was written by artist Y”)

  

  After you bring up the demo, use Google Chrome to navigate to Control Center at http://localhost:9021. Under MANAGEMENT select Topics, then scroll down and click on the topic and select Inspect.

  

  Inspect the topic play-events:

  

  Or use the KSQL query editor in Control Center to inspect the topic PRINT "play-events";:

  Inspect the topic song-feed by printing it with the FROM BEGINNING clause from the KSQL query editor in Control Center:

  PRINT "song-feed" FROM BEGINNING;
  
  

  Note: At this time, you cannot inspect data in the topic song-feed using the topic inspection capability in Control Center because topic inspection only works on new data, not previous data already produced to the topic, as is the case here.

  To review, the source data play-events and song-feed looks as follows:

  

  Create a new stream

  Let’s create a new stream from the Kafka topic, beginning with the play-events topic. Register the topic play-events as a KSQL stream and specify that it is Avro format.

  • Configure the new stream name as ksql_playevents
  • Change the message value encoding to AVRO (default: JSON)

  Note: Prefix the names of the KSQL streams and tables with ksql_. This is not required, but do it so that you can run these KSQL queries alongside the Kafka Streams API version of this music demo and avoid naming conflicts.

  Register the topic play-events from the KSQL query editor:

  CREATE STREAM ksql_playevents WITH (KAFKA_TOPIC='play-events', VALUE_FORMAT='AVRO')

  Or, from the KSQL Streams view, select “Create Stream” and fill out the fields as shown below. Because of Control Center integration with Confluent Schema Registry, it automatically detects the fields song_id and duration and their respective data types.

  

  Filter data

  Do some basic filtering on the newly created stream ksql_playevents, such as qualifying songs that were played for at least 30 seconds. This is what you get from the KSQL query editor:

  SELECT * FROM ksql_playevents WHERE DURATION > 30000;
  
  
  

  The above query is not persistent—it will stop if this screen is closed. To make the query persistent and stay running until explicitly terminated, prepend the previous query with CREATE STREAM <new stream name> AS. This is what you’ll see in the KSQL query editor:

  CREATE STREAM ksql_playevents_min_duration AS SELECT * FROM ksql_playevents WHERE DURATION > 30000;

  

  Now this persistent query will show in the queries list.

  Create a new table

  Next, let’s work on the song-feed topic, which effectively represents a table of songs. Data in a KSQL TABLE are required to have key of type String. In this demo, the table can be keyed on the song’s ID for joins and aggregations to work on that table.

  However the original Kafka topic has a key of type Long and the ID field is of type BIGINT. You can still create a TABLE with a few simple steps.

  1. Create a STREAM from the original Kafka topic song-feed:

  CREATE STREAM ksql_songfeed WITH (KAFKA_TOPIC='song-feed', VALUE_FORMAT='AVRO');

  Then DESCRIBE the stream to see the fields associated with this topic and notice that the field ID is of type BIGINT:

  DESCRIBE ksql_songfeed;

  

  2. Use the PARTITION BY clause to assign a key and use the CAST function to change the field type to String:

  CREATE STREAM ksql_songfeedwithkey WITH (KAFKA_TOPIC='KSQL_SONGFEEDWITHKEY', VALUE_FORMAT='AVRO') AS SELECT CAST(ID AS STRING) AS ID, ALBUM, ARTIST, NAME, GENRE FROM ksql_songfeed PARTITION BY ID;

  3. Convert the above stream into a TABLE with the ID field as its key, which is now of type String. This TABLE is a materialized view of events with only the latest value for each key, which represents an up-to-date table of songs

  CREATE TABLE ksql_songtable WITH (KAFKA_TOPIC='KSQL_SONGFEEDWITHKEY', VALUE_FORMAT='Avro', KEY='ID');

  Confirm that the entries in this KSQL table have a ROWKEY that matches the String ID of the song:

  SELECT * FROM ksql_songtable limit 5;

  Join play events with the table of songs

  At this point we have created a stream of filtered play events called ksql_playevents_min_duration as well as a table of song metadata called ksql_songtable.

  Enrich the stream of play events with song metadata using a stream-table JOIN. This will result in a new stream of play events enriched with descriptive song information like song title.

  CREATE STREAM ksql_songplays AS SELECT plays.SONG_ID AS ID, ALBUM, ARTIST, NAME, GENRE, DURATION, 1 AS KEYCOL FROM ksql_playevents_min_duration plays LEFT JOIN ksql_songtable songtable ON plays.SONG_ID = songtable.ID;

  Notice the addition of a clause 1 AS KEYCOL. For every row, this creates a new field KEYCOL that has a value of 1 ASKEYCOL can be later used in other derived streams and tables to do aggregations on a global basis.

  Create top music charts

  Now you can create a top music chart for all time to see which songs get played the most. Use the COUNT function on the stream ksql_songplays that we created above.

  CREATE TABLE ksql_songplaycounts AS SELECT ID, NAME, GENRE, KEYCOL, COUNT(*) AS COUNT FROM ksql_songplays GROUP BY ID, NAME, GENRE, KEYCOL;

  While the all-time greatest hits are cool, it would also be good to see statistics for just the last 30 seconds. Create another query, adding in a WINDOW clause, which gives counts of play events for all songs in 30-second intervals.

  CREATE TABLE ksql_songplaycounts30 AS SELECT ID, NAME, GENRE, KEYCOL, COUNT(*) AS COUNT FROM ksql_songplays WINDOW TUMBLING (size 30 seconds) GROUP BY ID, NAME, GENRE, KEYCOL;

  Here is what you built

  Congratulations, you built a streaming application that processes data in real time! The application enriched a stream of play events with song metadata and generated top counts. Any downstream systems can consume results from your KSQL queries for further processing. If you were already familiar with SQL semantics, hopefully this tutorial wasn’t too hard to follow.

  SELECT * FROM ksql_songplaycounts30 LIMIT 5;
  
  
  

  Learn more about what KSQL can do:

  • Download Confluent Platform and follow the quick start to get started with KSQL
  • Run the KSQL music application demo described in this tutorial and explore other automated demos showcasing KSQL
  • Watch the Level Up Your KSQL video series
  • Questions? Ask them in the #ksql channel on our community Slack group

  
  
  https://www.confluent.io/blog/building-streaming-application-ksql/