Apache Kafka News

• 

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

  1
  1
  Posts
  100
  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
  133
  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/
• 

  Streams and Tables: Two Sides of the Same Coin
  ksql stream processing kafka streams conferences streams • • sorangutan  

  1
  1
  Posts
  115
  Views

  

  We are happy to announce that our paper Streams and Tables: Two Sides of the Same Coin is published and available for free download. The paper was presented at the Twelfth International Workshop on Real-Time Business Intelligence and Analytics (BIRTE 2018) held in conjunction with the 44th International Conference on Very Large Data Bases (VLDB 2018) in Rio de Janeiro, Brazil, in August of this year.

  The BIRTE workshop attracted many participants and hosted a keynote, research, industry and demo session as well as a panel discussion about data stream processing.

  Paper Summary

  The paper is a joint work between Confluent and Humboldt-Universität zu Berlin that describes the Dual Streaming Model, which is the foundation of Kafka Streams’ and KSQL’s stream processing semantics:

  In this paper, we introduce the Dual Streaming Model to reason about physical and logical order in data stream processing. This model presents the result of an operator as a stream of successive updates, which induces a duality of results and streams. As such, it provides a natural way to cope with inconsistencies between the physical and logical order of streaming data in a continuous manner, without explicit buffering and reordering. We further discuss the trade-offs and challenges faced when implementing this model in terms of correctness, latency, and processing cost. A case study based on Apache Kafka illustrates the effectiveness of our model in the light of real-world requirements.
  Original source

  The Dual Streaming Model builds on the so-called stream-table duality, which allows you to unify data streams and relational tables into a holistic data processing model. Thus, data streams and continuously updating tables are the two core abstractions in the model. Additionally, the Dual Streaming Model decouples the handling of data that arrives later (i.e., out-of-order) from latency concerns and opens up a design space between processing cost, accepted latency and result completeness for the user that no other model offers.

  

  Figure 1. Design space

  The wide adoption and growth of Kafka Streams and KSQL among enterprises shows that the Dual Streaming Model solves real-world problems across all types of industries. As a result, we are elated to share our paper for free so you can become the stream processing expert in your company and take the business to the next level.

  Happy reading!

  Next Steps

  • Download the paper Streams and Tables: Two Sides of the Same Coin
  • View the slides from BIRTE 2018
  • Download Confluent Open Source to try out Kafka Streams and KSQL
  • Read the Kafka Streams and KSQL documentation
  • Check out this recommended read by Eno Thereska: Watermarks, Tables, Event Time, and the Dataflow Model

  
  
  https://www.confluent.io/blog/streams-tables-two-sides-same-coin
• 

  First-ever Certification Program for Apache Kafka
  • sorangutan  

  1
  1
  Posts
  212
  Views

  

  As Apache Kafka^® becomes the heart of the streaming data infrastructure at more and more organizations, we’ve been hearing from companies that they want some way of ensuring that the folks they hire are familiar with the platform, and are able to hit the ground running – as operators/administrators, or as developers. (Or both, of course.)

  Those requests led us to develop our new Certification program, which we launched today. You can now prove your Kafka knowledge by becoming a Confluent Certified Developer for Apache Kafka (CCDK) or Confluent Certified Administrator for Apache Kafka (CCAK). This will demonstrate that you have real Kafka knowledge: not just a basic grasp of the concepts, but a real, in-depth understanding of how things work, how to write resilient, reliable Kafka clients (CCDK), or how to manage a cluster, and diagnose and fix problems when they occur (CCAK).

  You will be able to take the exams at home or at your office, via an online proctored process – no need to travel to a test center and deal with a less-than-great environment. That also means the exams are available wherever you are in the world, something which is not always the case for certification programs that require in-person attendance. We are using a state-of-the-art proctoring company, for both your protection and ours; we want to ensure the integrity of the Certification process, and you as a candidate want to be sure that your credential is meaningful, and legitimate.

  The exams will be in-depth and detailed, but we’re not out to trick you. We expect that developers know the Java API, and that administrators are familiar with the major configuration parameters, but we won’t ask you the default value of that weird, obscure parameter that only five people have ever used in the history of the Kafka project. Our exams aim to test real-world knowledge and expertise, not whether or not you have a photographic memory.

  Our Certification exams are tied to a major release of Confluent Platform (such as version 3.x), and will be valid for two years from the date you take them. At that time, you’ll need to either recertify (if that major version is still active) or certify on the new active version. Why expire certifications? Because we know that an out-of-date Certification does no-one any good – Kafka is moving (and adding features) fast enough that the chances are it will have changed significantly two years from now. We know that not everyone upgrades Kafka immediately a new version is released, which is why we won’t expire or invalidate a Certification just because a new version of Confluent Platform is released. But employers, and companies hiring contractors, want to be sure that they people they’re hiring understand the platform as it exists now, not as it existed at some point in the dim and distant past.

  We’re announcing the Certification program now, and we will begin delivering the exams in April. In the meantime, watch https://www.confluent.io/training/ as we will be adding more information over the next few weeks, including study guides and sample questions. That’s also where you will find links to register for the exams as soon as they’re made live.

  The post First-ever Certification Program for Apache Kafka appeared first on Confluent.

  
  
  https://www.confluent.io/blog/first-ever-certification-program-for-apache-kafka/
• 

  Data Wrangling with Apache Kafka and KSQL
  ksql stream processing apache kafka kafka connect etl data wrangling • • sorangutan  

  1
  1
  Posts
  165
  Views

  

  KSQL, the SQL streaming engine for Apache Kafka^®, puts the power of stream processing into the hands of anyone who knows SQL. It’s fun to use for exploring data in Kafka topics, but its real power comes in building stream processing applications. By continually streaming messages from one Kafka topic to another, applying transformations expressed in SQL, it is possible to build powerful applications doing common data wrangling tasks such as:

  • Applying schema to data
  • Filtering and masking data
  • Changing data structures (for example, flattening nested data)
  • Changing the serialization format
  • Enriching streams of data
  • Unifying multiple streams of data

  Imagine you’ve got data on which you want to build some analytics. It’s coming from multiple sources—maybe different instances of the same system. You need to unify those sources into a single one, reformat the data along the way and apply a schema to the data so that it can be easily analyzed downstream. Apache Kafka and KSQL are a great choice for this.

  Why Kafka? Because we can use it for integrating the source systems with those downstream, in near real time and with the ability to add and swap out sources and targets without disturbing the rest of the components. In addition, we can reprocess data as required and share it with other systems—enabling them to take advantage of the data cleansing and wrangling that we’ve performed.

  

   

  In this article, we’ll see how to pull in data from REST sources, cleanse and wrangle it with KSQL, then stream it out to both Google Cloud Storage (GCS) as well as Google BigQuery for analysis and visualization in Google Data Studio. We’re using Confluent Cloud to host our Kafka brokers, but it will work on a local cluster instead if you want to.

  You can find all of the code used in this article on GitHub, including Docker Compose files.

  Getting the data in

  Credit: This section uses Environment Agency rainfall data from the real-time data API (beta).

  

   

  The data that I’m working with here is from the public API provided by the U.K. government’s Environment Agency. It contains readings taken from various weather stations around the country regarding information like rainfall, river levels and so on. Each station has its own REST API endpoint.

  We’re going to pull in data from four of these, using Kafka Connect to poll the sources. The data structures from each endpoint vary slightly, giving us a nice real-world example of data cleansing problems.

  I’m using the Kafka Connect REST source connector, which is a community-contributed connector written by Lenny Löfberg. Using this connector I can specify a REST endpoint and a polling interval, and the resulting payload is written to a Kafka topic.

  To start with, create the connectors (one per station):

  {
      "name": "source_rest_flood-monitoring-L2404",
      "config": {
          "connector.class": "com.tm.kafka.connect.rest.RestSourceConnector",
          "rest.source.url": "http://environment.data.gov.uk/flood-monitoring/id/stations/L2404",
          [...]
      }
  }

  → Full code on GitHub

  With this running, I have a Kafka topic per weather station and data in each:

  $ ccloud consume --from-beginning --topic flood-monitoring-059793|jq '.'                                                                                                                                 {
    "@context": "http://environment.data.gov.uk/flood-monitoring/meta/context.jsonld",
    "meta": {
      "publisher": "Environment Agency",
      [...]
      ]
    },
    "items": {
      [...]
      "label": "Rainfall station",
      "lat": 53.966652,
      "long": -1.115089,
      "measures": {
        [...]
        "latestReading": {
          "dateTime": "2018-08-22T05:30:00Z",
          "value": 0
          [...]

  You’ll note from the above sample of JSON that it’s nested data. For this station (059793) there is just one measure. However, for another station (3680) there are multiple measure readings, and the measure is supplied as an array:

  [...]
    "items": {
      "@id": "http://environment.data.gov.uk/flood-monitoring/id/stations/3680",
      "label": "Rainfall station",
      [...
      "measures": [
        {
          [...]
          "label": "rainfall-tipping_bucket_raingauge-t-15_min-mm",
          "latestReading": {
            "dateTime": "2018-08-30T04:00:00Z",
            "value": 0
            [...]
          },
        },
        {
          [...]
          "label": "temperature-dry_bulb-i-15_min-deg_C",
          "latestReading": {
            "dateTime": "2018-08-30T04:00:00Z",
            "value": 8
            [...]
          },
        }
      ],
  [...]

  This kind of varying structure in the data is a common problem in data integration and ETL. We’ll see in this article how KSQL can help with addressing this kind of challenge.

  Now, we want to take these four Kafka topics and build a stream processing application that will populate a single unified output topic of transformed data from across the four topics. This will process data that already exists in the Kafka topic, along with every new message that arrives.

  Declaring the schema

  The data coming in is JSON—but with no declared schema as such. As a framework, Kafka Connect can automagically register a schema for inbound data that it serializes as Apache Avro, but the REST connector here is basically just pulling string data from the REST endpoint, and that string data happens to be JSON. So, the first thing we’re going to do with KSQL is declare a schema for our source data from each topic. Note that the schema varies slightly to take into account the data from one of the stations that includes an array.

  

  CREATE STREAM flood_monitoring_059793 \
      (meta STRUCT<publisher         VARCHAR, \
                   comment           VARCHAR>, \
       items STRUCT<eaRegionName     VARCHAR, \
                    label            VARCHAR, \
                    stationReference VARCHAR, \
                    lat              DOUBLE, \
                    long             DOUBLE, \
                    measures STRUCT<label         VARCHAR, \
                                    latestReading STRUCT<dateTime VARCHAR, \
                                                         value DOUBLE>,\
                                    parameterName VARCHAR, \
                                    unitName VARCHAR>> \
      ) WITH (KAFKA_TOPIC='flood-monitoring-059793',VALUE_FORMAT='JSON');
  
  [...]
  
  CREATE STREAM flood_monitoring_3680 \
      (meta STRUCT<publisher VARCHAR, \
                   comment VARCHAR>, \
       items STRUCT<eaRegionName VARCHAR, \
                    label VARCHAR, \
                    stationReference VARCHAR, \
                    lat DOUBLE, \
                    long DOUBLE, \
                    measures ARRAY<STRUCT<label VARCHAR, \
                          latestReading STRUCT<\
                              dateTime VARCHAR, \
                              value DOUBLE>,\
                          parameterName VARCHAR, \
                          unitName VARCHAR>>> \
      ) WITH (KAFKA_TOPIC='flood-monitoring-3680',VALUE_FORMAT='JSON');

  With the Kafka topics registered and schemas defined, we can list them out:

  ksql> show streams;
  
   Stream Name             | Kafka Topic                 | Format
  ----------------------------------------------------------------
   FLOOD_MONITORING_3680   | flood-monitoring-3680       | JSON
   FLOOD_MONITORING_L2404  | flood-monitoring-L2404      | JSON
   FLOOD_MONITORING_059793 | flood-monitoring-059793     | JSON
   FLOOD_MONITORING_L2481  | flood-monitoring-L2481      | JSON
  ----------------------------------------------------------------

  Before we even do anything else to the data, we could use KSQL’s ability to reserialize and convert the raw JSON data into Avro. The advantage here is that any application downstream—whether it’s another KSQL process, Kafka Connect or a Kafka consumer—can work with the data directly from the topic and obtain the schema for it from the Confluent Schema Registry. To do this, use the CREATE STREAM…AS SELECT statement with the VALUE_FORMAT specified as part of the WITH clause:

  CREATE STREAM FLOOD_MONITORING_3680_AVRO \
      WITH (VALUE_FORMAT='AVRO') AS \
      SELECT * FROM FLOOD_MONITORING_3680;

  You can also define the partitioning and replication factor at this stage, too, if you wanted to change that.

  For the rest of this exercise we’ll stick to the original JSON topics and apply the Avro serialization later on.

  Working with nested data

  Because the data in the source topic is nested JSON, we declare the parent column’s data type as STRUCT. To access the data with KSQL, use the -> operator:

  select items->stationreference, \
         items->earegionname, \
         items->measures->parameterName, \
         items->measures->latestreading->datetime,\
         items->measures->latestreading->value, \
         items->measures->unitname \
  from   flood_monitoring_L2481 limit 1;
  L2481 | North East | Water Level | 2018-08-22T13:00:00Z | 5.447 | mAOD

  For the readings that are part of an array (declared as an ARRAY<STRUCT>), use square brackets to designate the index:

  ksql> select items->stationreference, \
          items->earegionname, \
          items->measures[0]->parameterName, \
          items->measures[0]->latestreading->datetime,\
          items->measures[0]->latestreading->value, \
          items->measures[0]->unitname \
   from   flood_monitoring_3680  limit 1;
  3680 | Midland | Rainfall | 2018-08-30T04:00:00Z | 0.0 | mm

  Unifying data from multiple streams

  Taking a sample record from each topic and reading type gives us this table when manually collated:

  Station reference	Station region	Measurement type	Timestamp	Measurement	Unit
  3680	Midland	Rainfall	2018-08-30T04:00:00Z	0.0	mm
  3680	Midland	Temperature	2018-08-30T04:00:00Z	8.0	deg C
  059793	North East	Rainfall	2018-08-22T05:30:00Z	0.0	mm
  L2481	North East	Water Level	2018-08-22T13:00:00Z	5.447	mAOD
  L2404	North East	Water Level	2018-08-22T18:45:00Z	5.23	mAOD

  Looking at the data, we can apply a data model that looks something like this:

  +-Environment Readings
    +-Station reference
    +-Station region
    +-Type of measurement
    +-Measurement timestamp
    +-Measurement
    +-Measurement units

  The type of reading (e.g., temperature, rainfall and river level) varies, as do the units associated with it and the station. We could normalize this out into stations, reading types and so on—but for ease of reporting we’ll actually denormalize it into a single flat structure. This means bringing in data from the multiple streams, including manually exploding the array within the 3680 topic (which contains both rainfall and temperature data):

  

   

  To do this, we can use KSQL’s INSERT INTO statement. This streams the results of a SELECT statement into an existing target STREAM. We’ll create the initial STREAM using CREATE STREAM…AS SELECT. We’ll also take the opportunity to serialize the data to Avro.

  CREATE STREAM ENVIRONMENT_DATA WITH \
          (VALUE_FORMAT='AVRO') AS \
  SELECT  items->stationreference AS stationreference, \
          items->earegionname AS earegionname, \
          items->label AS label, \
          items->lat AS lat, items->long AS long, \
          items->measures->latestreading->datetime AS reading_ts, \
          items->measures->parameterName AS parameterName, \
          items->measures->latestreading->value AS reading_value, \
          items->measures->unitname AS unitname \
   FROM   flood_monitoring_L2404 ;
  
  INSERT INTO ENVIRONMENT_DATA \
  SELECT  items->stationreference AS stationreference, \
          items->earegionname AS earegionname, \
          items->label AS label, \
          items->lat AS lat, items->long AS long, \
          items->measures->latestreading->datetime AS reading_ts, \
          items->measures->parameterName AS parameterName, \
          items->measures->latestreading->value AS value, \
          items->measures->unitname AS unitname \
   FROM   flood_monitoring_L2481 ;
  
  -- (INSERT INTO repeated for the remaining source topics)

  → Full code on GitHub

  Now when we inspect the STREAMS, we can see the new one created and populated by the above statements:

  ksql> show streams;
  
   Stream Name                | Kafka Topic                 | Format
  -------------------------------------------------------------------
   FLOOD_MONITORING_3680      | flood-monitoring-3680       | JSON
   FLOOD_MONITORING_L2404     | flood-monitoring-L2404      | JSON
   FLOOD_MONITORING_059793    | flood-monitoring-059793     | JSON
   FLOOD_MONITORING_L2481     | flood-monitoring-L2481      | JSON
   ENVIRONMENT_DATA           | ENVIRONMENT_DATA            | AVRO
  -------------------------------------------------------------------

  Note that the “format” as shown in the column in the above output is AVRO. Using DESCRIBE EXTENDED you can check that messages are being processed by reviewing the Local runtime statistics:

  ksql> DESCRIBE EXTENDED ENVIRONMENT_DATA;
  [...]
  Local runtime statistics
  ------------------------
  messages-per-sec:         0   total-messages:      2311     last-message: 8/30/18 2:38:48 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 ENVIRONMENT_DATA)
  ksql>

  The unified topic is ENVIRONMENT_DATA and has data from all source topics within it:

  ksql> SELECT * FROM ENVIRONMENT_DATA ;
  1534992115367 | null | L2404 | North East | Foss Barrier | 53.952443 | -1.078056 | 2018-08-22T18:45:00Z | Water Level | 5.23 | mAOD
  [...]
  1535615911999 | null | L2481 | North East | York James Street TS | 53.960145 | -1.06865 | 2018-08-30T05:30:00Z | Water Level | 5.428 | mAOD
  [...]
  1535135263726 | null | 059793 | North East | Rainfall station | 53.966652 | -1.115089 | 2018-08-24T17:00:00Z | Rainfall | 0.0 | mm
  [...]
  1535638518251 | null | 3680 | Midland | Rainfall station | 52.73152 | -0.995167 | 2018-08-30T04:00:00Z | Rainfall | 0.0 | mm
  [...]
  1535638518251 | null | 3680 | Midland | Rainfall station | 52.73152 | -0.995167 | 2018-08-30T04:00:00Z | Temperature | 8.0 | deg C

  Re-keying data in KSQL

  Based on the above data model, the unique key for data is a composite of the station, reading type and timestamp. We’re going to handle the timestamp separately—for now, let’s see how to use KSQL to set the message key used by Kafka.

  The message key is important as it defines the partition on which messages are stored in Kafka and is used in any KSQL joins. At the moment there’s no key set, so data for the same station and reading type could be scattered across partitions. For a few rows of data this may not matter, but as volumes increase it becomes more important to consider. It’s also pertinent to the strict ordering guarantee that Kafka provides, which only applies within a partition.

  Using kafkacat we can inspect the partition assignments. I’m using a topic that I’ve created just for this purpose, with the serialization set to JSON (kafkacat doesn’t currently support Avro). By filtering for a given station we can see the partitions that messages are assigned to as well as the message key:

  $ kafkacat -b kafka-broker:9092 -t ENVIRONMENT_DATA_JSON -f 'Partition: %p\tOffset: %o\tKey (%K bytes): %k\tValue (%S bytes): %s\n'|grep L2481
  
  Partition: 0    Offset: 344 Key (-1 bytes):  Value (260 bytes): {"STATIONREFERENCE":"L2481"[...]
  [...]
  Partition: 1    Offset: 595 Key (-1 bytes):  Value (260 bytes): {"STATIONREFERENCE":"L2481"[...]
  [...]
  Partition: 2    Offset: 48  Key (-1 bytes):  Value (260 bytes): {"STATIONREFERENCE":"L2481"[...]
  Partition: 2    Offset: 49  Key (-1 bytes):  Value (260 bytes): {"STATIONREFERENCE":"L2481"[...]
  [...]

  Note that the messages span several partitions and have a null key.

  Next, let’s repartition our unified data stream using the PARTITION BY clause:

  CREATE STREAM ENVIRONMENT_DATA_REKEYED AS \
      SELECT STATIONREFERENCE+PARAMETERNAME AS COMPOSITE_KEY, * FROM ENVIRONMENT_DATA \
      PARTITION BY COMPOSITE_KEY;

  Checking the data with kafkacat again, we see the following:

  kafkacat -b kafka-broker:9092 -t ENVIRONMENT_DATA_REKEYED -f 'Partition: %p\tOffset: %o\tKey (%K bytes): %k\tValue (%S bytes): %s\n'|grep L2481
  % Auto-selecting Consumer mode (use -P or -C to override)
  % Reached end of topic ENVIRONMENT_DATA_REKEYED2 [3] at offset 0
  % Reached end of topic ENVIRONMENT_DATA_REKEYED2 [1] at offset 0
  Partition: 2    Offset: 0       Key (16 bytes): L2481Water Level        Value (241 bytes): {"COMPOSITE_KEY":"L2481Water Level","STATIONREFERENCE":"L2481"[...]
  Partition: 2    Offset: 1       Key (16 bytes): L2481Water Level        Value (241 bytes): {"COMPOSITE_KEY":"L2481Water Level","STATIONREFERENCE":"L2481"[...]
  [...]
  Partition: 2    Offset: 734     Key (16 bytes): L2481Water Level        Value (241 bytes): {"COMPOSITE_KEY":"L2481Water Level","STATIONREFERENCE":"L2481"[...]

  All of the messages for the given key reside on a single partition, and each message has a key and value.

  Managing message timestamps in KSQL

  As well as messages having a key (and value), they also have a timestamp in their metadata. This can be set explicitly by the application producing the messages to Kafka, or in the absence of that messages will take the time at which it arrives at the Kafka broker. The timestamp of the messages we’re working with have the timestamp of the time at which they were ingested by Kafka Connect. However, the actual timestamp to use in processing the data for analysis is the items.measures.latestReading.dateTime value within the message. This matters particularly when using the data for aggregations, time-based partitioning and so on. Using the TIMESTAMPTOSTRING function we can examine the two timestamps discussed above:

  ksql> SELECT TIMESTAMPTOSTRING(ROWTIME, 'yyyy-MM-dd HH:mm:ss'), items->measures->latestReading->dateTime FROM FLOOD_MONITORING_L2404 LIMIT 1;
  2018-08-23 01:11:53 | 2018-08-22T18:45:00Z

  In this example, the data arrived in Kafka at 01:11 on Aug. 23, 2018, but the reading occurred at 18:45 on the Aug. 22, 2018. If we did any arithmetic on the data as it stands (for example, calculating the maximum reading value on Aug. 22), we’d get an incorrect answer. This is because KSQL uses the message timestamp (accessible through the virtual system column ROWTIME) in its time processing (such as windowed aggregates).

  To rectify this, we can use KSQL. Just as we used the WITH clause previously to set the serialization format to Avro, we can use a similar pattern to override the timestamp that will be used for the messages in the target stream being created:

  CREATE STREAM ENVIRONMENT_DATA_WITH_TS \
              WITH (TIMESTAMP='READING_TS', \
                    TIMESTAMP_FORMAT='yyyy-MM-dd''T''HH:mm:ssX') AS \
  SELECT * \
    FROM ENVIRONMENT_DATA ;

  This is currently blocked by #1439. The workaround is a two-step conversion: CREATE STREAM ENVIRONMENT_DATA_WITH_TS_STG AS \ SELECT STRINGTOTIMESTAMP(READING_TS, 'yyyy-MM-dd''T''HH:mm:ssX') AS READING_TS_EPOCH, * \ FROM ENVIRONMENT_DATA ; CREATE STREAM ENVIRONMENT_DATA_WITH_TS \ WITH (TIMESTAMP='READING_TS_EPOCH') AS \ SELECT * \ FROM ENVIRONMENT_DATA_WITH_TS_STG;

  You can validate the conversion by comparing the ROWTIME of the newly created stream with the source READING_TS:

  ksql> SELECT TIMESTAMPTOSTRING(ROWTIME, 'yyyy-MM-dd HH:mm:ss'), READING_TS \
        FROM ENVIRONMENT_DATA_WITH_TS LIMIT 1;
  2018-08-22 18:45:00 | 2018-08-22T18:45:00Z

  Column derivations with KSQL

  Beyond filtering, KSQL can be used to create derivations based on the incoming data. Let’s take the example of dates. As well as the raw timestamp of each reading that we receive, it could be that for ease of use downstream we want to also add columns for just year, month and so on. The TIMESTAMPTOSTRING function and DateTime format strings allow you to easily accomplish these tasks:

  CREATE STREAM ENVIRONMENT_DATA_LOCAL_WITH_TS_AND_DATE_COLS AS \
  SELECT *, \
         TIMESTAMPTOSTRING(ROWTIME,'QQQ') as READING_QTR, \
         TIMESTAMPTOSTRING(ROWTIME,'yyyy-MM-dd') as READING_YMD, \
         TIMESTAMPTOSTRING(ROWTIME,'yyyy-MM') as READING_YM \
  FROM   ENVIRONMENT_DATA_LOCAL_WITH_TS;

  The stream now includes the new columns:

  ksql> DESCRIBE ENVIRONMENT_DATA_LOCAL_WITH_TS_AND_DATE_COLS;
  
  Name                 : ENVIRONMENT_DATA_LOCAL_WITH_TS_AND_DATE_COLS
   Field            | Type
  ----------------------------------------------
   ROWTIME          | BIGINT           (system)
   ROWKEY           | VARCHAR(STRING)  (system)
   READING_TS_EPOCH | BIGINT
   STATIONREFERENCE | VARCHAR(STRING)
   EAREGIONNAME     | VARCHAR(STRING)
   LABEL            | VARCHAR(STRING)
   LAT              | DOUBLE
   LONG             | DOUBLE
   READING_TS       | VARCHAR(STRING)
   PARAMETERNAME    | VARCHAR(STRING)
   READING_VALUE    | DOUBLE
   UNITNAME         | VARCHAR(STRING)
   READING_QTR      | VARCHAR(STRING)
   READING_YMD      | VARCHAR(STRING)
   READING_YM       | VARCHAR(STRING)
  ----------------------------------------------

  Each message includes the derived data:

  ksql> SELECT READING_TS, \
               READING_QTR, \
               READING_YMD, \
               READING_YM \
        FROM   ENVIRONMENT_DATA_LOCAL_WITH_TS_AND_DATE_COLS \
               LIMIT 5;
  2018-08-26T18:15:00Z | Q3 | 2018-08-26 | 2018-08
  2018-08-24T18:15:00Z | Q3 | 2018-08-24 | 2018-08
  2018-08-29T09:15:00Z | Q3 | 2018-08-29 | 2018-08
  2018-08-23T18:15:00Z | Q3 | 2018-08-23 | 2018-08
  2018-08-25T05:30:00Z | Q3 | 2018-08-25 | 2018-08
  Limit Reached
  Query terminated

  Note: In this example we’re building step by step a series of transformations in a daisy-chain style. In practice you may refactor them into fewer steps, but I’m keeping them separate here to make the explanations clearer.

  Filtering data with KSQL

  Let’s see how we can filter the data using KSQL. Each CREATE STREAM…AS SELECT statement creates a Kafka topic populated continually with the results of the transformation. We can use the same approach to filter the stream of data. For example, using the KSQL function GEO_DISTANCE, we can filter the stream of data to just messages within the given distance of a point.

  CREATE STREAM ENVIRONMENT_DATA_LOCAL_WITH_TS AS \
  SELECT * FROM ENVIRONMENT_DATA_WITH_TS \
  WHERE  GEO_DISTANCE(LAT,LONG,53.919066, -1.815725,'KM') < 100;

  Masking data with KSQL

  As well as filtering entire rows based on a predicate as shown above, KSQL can also be used to “filter” columns from a message. Imagine you have a field in your source data that you don’t want to persist downstream—with KSQL you simply create a derived stream and omit the column(s) in question from the projection:

  ksql> DESCRIBE ENVIRONMENT_DATA;
  
  Name                 : ENVIRONMENT_DATA
   Field            | Type
  ----------------------------------------------
   ROWTIME          | BIGINT           (system)
   ROWKEY           | VARCHAR(STRING)  (system)
   STATIONREFERENCE | VARCHAR(STRING)
   EAREGIONNAME     | VARCHAR(STRING)
   LABEL            | VARCHAR(STRING)
   LAT              | DOUBLE
   LONG             | DOUBLE
   READING_TS       | VARCHAR(STRING)
   PARAMETERNAME    | VARCHAR(STRING)
   READING_VALUE    | DOUBLE
   UNITNAME         | VARCHAR(STRING)
  ----------------------------------------------
  
  ksql> CREATE STREAM ENVIRONMENT_DATA_MINIMAL AS \
          SELECT STATIONREFERENCE, READING_TS, READING_VALUE \
          FROM ENVIRONMENT_DATA;
  
  ksql> DESCRIBE ENVIRONMENT_DATA_MINIMAL;
  
  Name                 : ENVIRONMENT_DATA_MINIMAL
   Field            | Type
  ----------------------------------------------
   ROWTIME          | BIGINT           (system)
   ROWKEY           | VARCHAR(STRING)  (system)
   STATIONREFERENCE | VARCHAR(STRING)
   READING_TS       | VARCHAR(STRING)
   READING_VALUE    | DOUBLE
  ----------------------------------------------

  As well as simply dropping a column, KSQL ships with functions to mask data:

  ksql> SELECT STATIONREFERENCE, EAREGIONNAME \
        FROM ENVIRONMENT_DATA;
  L2404 | North East
  
  ksql> SELECT STATIONREFERENCE, EAREGIONNAME, MASK_RIGHT(EAREGIONNAME,4) AS REGION_NAME_MASKED \
        FROM ENVIRONMENT_DATA2;
  L2404 | North East | North Xxxx

  There are several MASK-based functions, and if you have your own special sauce you’d like to use here, KSQL does support user-defined functions (UDFs) as of Confluent Platform 5.0.

  Recap

  So far, we’ve ingested data from several sources with similar but varying data models. Using KSQL, we’ve wrangled the data by:

  • Flattening nested data structures
  • Reserializing JSON data to Avro
  • Unifying the multiple streams into one
  • Setting the message partitioning key
  • Setting the message timestamp metadata to the correct logical value
  • Creating derived columns in the transformation
  • Filtering and masking the data

  

  The results of these transformations is continually populated Kafka topics. As new messages arrive on the source, continuously running KSQL statements process and write them to the target Kafka topic.

  Streaming onwards …

  The great thing about Kafka is its ability to build systems in which functionality is compartmentalized. Ingest is handled by one process (in this case, Kafka Connect), and transformation is handled by a series of KSQL statements. Each can be modified and switched out for another without impacting the pipeline we’re building. Keeping them separate makes it easier to perform important activities such as testing, troubleshooting and analyzing performance metrics. It also means that we can extend data pipelines easily.

  We may have a single use case in mind when initially building it, and one way to do this would be building a single application that pulls data from REST endpoints before cleansing, wrangling and writing it out to the original target. But now if we want to add other targets, we have to modify that application, which becomes more complex and risky. Instead, by breaking up the processes and building them all around Kafka, adding another target for the data is as simple as consuming the transformed data from a Kafka topic.

  So, let’s take our transformed data and do something with it! We can use it to drive analytic requirements, but we’ll also see how it can drive applications themselves, too.

  

  For our analytics, we’re going to land the data to BigQuery, Google’s cloud data warehouse tool. We’ll use another Kafka Connect community connector, one written by WePay to stream data from Kafka topics to BigQuery. You’ll need to set up your Google Cloud Platform (GCP) credentials in a file accessible to the Connect worker(s), and also make sure that the BigQuery project and dataset exist first. Here, I’m using ones called devx-testing and environment_data, respectively:

  {
    "name": "sink_gbq_environment-data",
    "config": {
      "connector.class":"com.wepay.kafka.connect.bigquery.BigQuerySinkConnector",
      "topics": "ENVIRONMENT_DATA",
      "autoCreateTables":"true",
      "autoUpdateSchemas":"true",
      "project":"devx-testing",
      "datasets":".*=environment_data",
      "keyfile":"/root/creds/gcp_creds.json"
      [...]

  → Full code on GitHub

  Once deployed, we can see data arriving in BigQuery using the GCP console:

  

  We can also use bq:

  $ bq ls environment_data
           tableId           Type    Labels   Time Partitioning
   ------------------------ ------- -------- -------------------
    ENVIRONMENT_DATA         TABLE            DAY
  
  $ bq query 'select * from environment_data.ENVIRONMENT_DATA'
  Waiting on bqjob_r5ce1258159e7bf44_000001658f8cfedb_1 ... (0s) Current status: DONE
  +------------------+--------------+------------------------+------+------------+----------------------+-----------+-----------+----------------------+---------------+-------+----------+
  | STATIONREFERENCE | EAREGIONNAME |       EAAREANAME       | TOWN | RIVERNAME  |        LABEL         |    LAT    |   LONG    |       DATETIME       | PARAMETERNAME | VALUE | UNITNAME |
  +------------------+--------------+------------------------+------+------------+----------------------+-----------+-----------+----------------------+---------------+-------+----------+
  | L2404            | North East   | North East - Yorkshire | York | River Ouse | Foss Barrier         | 53.952443 | -1.078056 | 2018-08-08T16:30:00Z | Water Level   |  5.01 | mAOD     |
  | L2404            | North East   | North East - Yorkshire | York | River Ouse | Foss Barrier         | 53.952443 | -1.078056 | 2018-08-08T18:15:00Z | Water Level   | 5.003 | mAOD     |
  [...]

  There are many ways to work with data in BigQuery: the direct SQL interface, the GUI console—or through numerous analytics visualization tools, including Looker, Tableau, Qlik, Redash, etc. Here, I’ve used Google’s own Data Studio. Connecting to BigQuery is simple, and once the dataset is in Data Studio, it’s a matter of moments to throw some useful visualizations together:

  

   

  We’ve discussed streaming data to Google BigQuery, but did you know that you can also stream the same transformed data to GCS for archival purposes or even batch access from other applications (although arguably this would be done from consuming the Kafka topic directly)?

  {
    "name": "sink_gcs_environment-data",
    "config": {
      "connector.class": "io.confluent.connect.gcs.GcsSinkConnector",
      "topics": "ENVIRONMENT_DATA",
      "gcs.bucket.name": "rmoff-environment-data",
      "gcs.part.size": "5242880",
      "flush.size": "16",
      "gcs.credentials.path": "/root/creds/gcp_creds.json",
  [...]

  → Full code on GitHub

  With this connector running, we now have data streaming to both BigQuery and GCS:

  $ gsutil ls gs://rmoff-environment-data/topics/
  gs://rmoff-environment-data/topics/ENVIRONMENT_DATA/
  
  $ gsutil ls gs://rmoff-environment-data/topics/ENVIRONMENT_DATA/partition=0/
  gs://rmoff-environment-data/topics/ENVIRONMENT_DATA/partition=0/ENVIRONMENT_DATA+0+0000000000.json
  gs://rmoff-environment-data/topics/ENVIRONMENT_DATA/partition=0/ENVIRONMENT_DATA+0+0000000016.json

  Conclusion

  KSQL and Apache Kafka are a powerful way to build integration between systems, with transformation applied to the data in flight, and the resulting data available to multiple consuming applications downstream. By working with streaming data, organizations can take advantage of the transform-once-use-many paradigm, since the data is available instantly for real-time applications to use. Applications with less immediate requirements (as is often the case with analytics) can still use the same data, reducing complexity, minimizing the duplication of code and leading to a more flexible and powerful architecture. 

  Using KSQL, streaming processing can be expressed using SQL alone so you don’t need to write any Java code—opening up stream processing to a much wider audience of developers.

  If you’re interested in what KSQL can do:

  • Watch the introductory video of 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

  You can find all of the code used in this article on GitHub.

  The post Data Wrangling with Apache Kafka and KSQL appeared first on Confluent.

  
  
  https://www.confluent.io/blog/data-wrangling-apache-kafka-ksql
• 

  Putting the Power of Apache Kafka into the Hands of Data Scientists
  event apache kafka architecture integration self-service • • sorangutan  

  1
  1
  Posts
  153
  Views

  

  The following post is by guest blogger Liz Bennett, a member of the Compute Infrastructure Team at Stitch Fix, where she has been leading efforts to expand Stitch Fix’s logging and data integration capabilities. You can find the original post on MultiThreaded.

  At Stitch Fix, we blend the art of expert personal styling with the science of algorithms to deliver highly personalized services that cater to the tastes, lifestyles and budgets of our 2.7 million active clients. As the world’s leading online personal styling service, Stitch Fix relies heavily on efficient data collection and analysis in order to sustain and grow the business.

  Over a year ago, my fellow data infrastructure engineers and I broke ground with a total rewrite of our event delivery infrastructure. Our mission was to build a robust, centralized data integration platform tailored to the needs of our Data Scientists. The platform would be fully self-service, so as to maximize the Data Scientists’ autonomy and give them complete control over their event data. Ultimately, we delivered a platform that is revolutionizing the way Data Scientists interact with Stitch Fix’s data.

  In two parts, this post peeks into Stitch Fix’s data science culture and delves into how it drove the fundamental decisions we made in our lowest levels of data infrastructure. Part 1 discusses our design process, explains our guiding philosophy around self-service tooling and explores our data integration platform concept. Part 2 is a technical dive into the decisions we made and a walkthrough of the whole architecture.

  Part 1: The Design

  Why build event delivery infrastructure

  Around the time we started working on our event delivery rewrite, much of the data in our data warehouse was being collected from database snapshots. This style of data pipeline has been ubiquitous across the industry for decades, as database snapshots prove useful in many ways, are generally easy to set up and require low maintenance. They have a few drawbacks, however:

  • You don’t know how the database’s state changes in between snapshots.
  • Data can only be analyzed once a snapshot is taken—analysis cannot happen in real time.
  • Context and metadata about the state is often missing. Only data that exists in the online transaction processing (OLTP) database can be analyzed (i.e., the state that is necessary for operational applications).

  But events could solve these problems. Building a robust, centralized and scalable system dedicated to collecting event data would greatly augment the database snapshot data and unlock loads of potential for Data Scientists.

  Self-service event delivery

  Our main requirement for this new project was to build infrastructure that would be 100 percent self-service for our Data Scientists. In other words, my teammates and I would never be directly involved in the discovery, creation, configuration and management of the event data. Self-service would fix the primary shortcoming of our legacy event delivery system: manual administration that was performed by my team whenever a new dataset was born. This manual process hindered the productivity and access to event data for our Data Scientists. Meanwhile, fulfilling the requests of the Data Scientists hindered our own ability to improve the infrastructure. This scenario is exactly what the Data Platform Team strives to avoid. Building self-service tooling is the number one tenet of the Data Platform Team at Stitch Fix, so whatever we built to replace the old event infrastructure needed to be self-service for our Data Scientists. You can learn more about our philosophy in Jeff Magnusson’s post Engineers Shouldn’t Write ETL.

  Making the event delivery infrastructure completely self-service to Data Scientists was a unique and interesting challenge. Having spent years working on real-time streaming infrastructure prior to Stitch Fix, I knew how to craft efficient Kafka applications, monitor them, scale them and maintain a Kafka cluster in production. But how could we, a lean team of three, not only deploy our own brand-new Kafka cluster but also design and build a self-service event delivery platform on top of it? How could we give Data Scientists total control and freedom over Stitch Fix’s event data without requiring them to understand the intricacies of streaming data pipelines?

  Designing the data integration platform

  Designing our self-service layer on top of Kafka began with an extensive period of research. Noteworthy inspirations for our data integration platform concept came from Martin Kleppmann’s Designing Data-Intensive Applications: The Big Ideas Behind Reliable, Scalable, and Maintainable Systems and Apache Kafka^® Co-creator Jay Kreps’ article The Log: What Every Software Engineer Should Know About Real-Time Data’s Unifying Abstraction.

  In the article, Kreps talks about the power of a centralized log pipeline and at one point focuses specifically on the simplification of data integration, where he provides this definition:

  “Data integration is making all the data an organization has available in all its services and systems.”

  He goes on to say the following, which I found extremely compelling: 

  “You don’t hear much about data integration in all the breathless interest and hype around the idea of big data, but nonetheless, I believe this mundane problem of ‘making the data available’ is one of the more valuable things an organization can focus on.”

  I was enamored by this idea of “making all the data an organization has available in all its services and systems.” What could be more powerful than giving Data Scientists easy, self-service access to all data across the whole company?

  To explain how this would all work, I’ll paraphrase Kreps’ idea of data integration and how it solves the problem of making data available everywhere. First, picture all the services, applications, specialized data systems and databases within a company. As the company evolves, use cases pop up that involve connecting one thing to another in order to pass data back and forth. The teams building these connections likely build a one-off dedicated solution, because it is quicker and easier in the short term than building a general purpose solution. When enough teams perform this, an architecture such as the one shown in Figure 1 materializes.

  

  Figure 1. Data integration without a centralized log pipeline

  Each arrow is a custom connection that has to be individually developed and maintained. To achieve total connectivity in which every process can stream data back and forth between every other process within the company, you’d have to build n² connectors, where n is the total number of processes within the company. Any engineer will tell you that this is unscalable.

  Now imagine the same set of services, applications, specialized data systems and databases with the addition of a centralized log pipeline. Instead of passing data directly between processes, the data flows through the centralized pipeline, yielding an architecture as shown in Figure 2.

  

  Figure 2. Data integration with a centralized log pipeline

  This approach reduces connections by an order of magnitude and accomplishes the same goal—data can flow from any process to any other process within the company. The beauty of this architecture is that every connector is quite uniform in function, which allows for a general purpose platform that handles getting data in and out of the central pipeline. This platform centralizes and simplifies the mechanisms by which data is transported from A to B, making everyone’s job much easier.

  With that in mind, we planned for this general purpose platform to become our data integration platform. Each connector would be code written, deployed and maintained by my teammates. We would build a simple web based user interface (UI) for Data Scientists to configure their connectors, and easily get data in and out of the central log pipeline. Inspired by the network of roads and highways that connect everything in the world, we decided to call it The Data Highway.

  

  Part 2: Building

  With a clear vision of what to build, we entered another phase of research, this time focused on comparing specific technologies. Because the data infrastructure world is still blossoming into a mature ecosystem, there were many young technologies to choose from. In this section, I’ll outline the reasons we chose the solutions that we did.

  Choosing a message broker

  Currently, the most frequently adopted solutions for high-volume, distributed message brokers are Apache Kafka, Amazon Kinesis and Cloud Pub/Sub. We ruled out Cloud Pub/Sub early on simply because integrating our Amazon Web Services (AWS) stack with a Google Cloud Platform (GCP) product would be too costly. That left Kinesis vs. Kafka.

  I could devote an entire post to this decision alone, because there were a multitude of factors we considered, many of which were nuanced and difficult to measure objectively. Our data volume is modest enough that either Kafka or Kinesis would have worked purely from a performance perspective. Hence, our decision-making focused on questions like, “Which technology will be easier to write tooling for? Which technology will give us a better foundation to build our platform on top of? Which technology will be the easiest to maintain and evolve?” Answering these questions would have been very difficult were it not for the team’s collective experience using both technologies.

  A year prior to starting this project, we had stood up a basic event delivery system using Kinesis and AWS Lambda. The system was running decently well—especially considering how easy it was to spin up—but we were quickly outgrowing it for a number of reasons:

  • A lot of manual setup was required for new datasets.
  • Maintaining separate staging and production environments was tricky due to the way that the Lambda functions were set up.
  • We had to make some architectural tradeoffs to work around Kinesis’ throttling policies.
  • Worst of all, the system had grown too complex to be self-service for Data Scientists.

  Ultimately, we had to decide whether to re-architect our existing Kinesis system or start from scratch with a Kafka-based system.

  After some debate, we chose Kafka. Although the ease of a hosted solution like Kinesis was very tempting, the scales tipped towards Kafka because:

  • We wanted something that would integrate well with AWS and non-AWS products alike.
  • We wanted fine-grained control over our deployment.
  • An in-house deployment of Kafka would scale more than Kinesis and be cheaper in the long run.
  • We wanted to improve our involvement in the open source community.
  • The Kafka ecosystem is richer and more fully featured than the Kinesis ecosystem.
  • As far as databases go, Kafka has a relatively low operational burden.

  Comparing data integration solutions

  For the data integration layer on top of Kafka, we considered lots of options. There is a budding ecosystem for these kinds of tools in the open source world, so we wanted to use an existing technology rather than write our own system. The primary contenders were Apache Beam, Apache Storm, Logstash, and Kafka Connect.

  Apache Beam is a Swiss Army knife for building batch and streaming ETL pipelines. Its Pipeline I/O feature comes with a suite of built-in read and write transforms. These transforms read and write data from a variety of places, and there is ongoing work to add many more options for consuming and producing data. Apache Beam was not designed to be a data integration solution, however, although it has some data integration capabilities. It is a flexible data processing software development kit (SDK), not a specialized system for getting data from A to B. We would have had to write a ton of code to support fully self-service data pipelines that are configurable with a click of a button.

  Apache Storm was a potential option, but it would have been costly to build the self-service interface on top of it, as the data flow pipelines have to be written as code and deployed yourself. Apache Storm is better suited to use cases where you have a very specialized streaming pipeline with clearly defined requirements. We needed something that was multipurpose and would be easily configurable even in a runtime environment.

  Logstash is handy because it has so many integrations, more than that of any other options we researched. It’s also lightweight and easy to deploy. Yet Logstash lacked important resiliency features, such as support for distributed clusters of workers and disk-based buffering of events. As of version 6.0, both of these features are now implemented, so Logstash could work for a data integration layer. But its biggest drawback for our use case is that the data flows are configured using static files written in Logstash’s (somewhat finicky) DSL. This would make it challenging to build the self-service interface. (Note: We did actually end up using Logstash to ingest Syslog data—more on that later.)

  Finally, we investigated Kafka Connect. Kafka Connect’s goal is to make the integration of systems as simple and resilient as possible. It is built on top of the standard Kafka consumer and producer, so it has automatic load balancing. It’s simple to adjust processing capacity, and it has strong delivery guarantees. It can run in a distributed cluster so failovers, as well as scaling and rolling restarts, are seamless. It is stateless because it stores all its state in Kafka itself. In addition, there is a growing array of open source connectors that can be downloaded and installed into your deployment. If you have special requirements, the API for writing your own connectors is simple and expressive, making it straightforward to write custom integrations. The most compelling feature for us was Kafka Connect’s REST API for creating, deleting, modifying or pausing connectors in a runtime environment. This provided a super-easy way to build our self-service interface.

  It’s not surprising that Kafka Connect was the best solution for our needs. The inspiration for our data integration platform came primarily from the creators of Kafka. They designed Kafka Connect specifically to help bring their vision to life and solve the often underestimated challenge of simplifying data integration. I can confirm that Kafka Connect delivers. It is a supremely well-engineered piece of technology that is the workhorse of our whole platform. It saved us a tremendous amount of effort, and it is getting better every day as more integrations are being added and its features mature. It is well-maintained, and we were able to get a few patches merged upstream without much ado. It’s not totally perfect—it has some painful classpath problems—but overall we’re very happy with it.

  Putting all the pieces together

  Now for the moment you’ve all been waiting for … the architecture diagram!

  

  Figure 3. The Data Highway complete architecture diagram

  For the most part, we were able to leverage open-source technologies to power our platform, but there were a handful of things that we wrote ourselves to fill in the gaps, which I’ll briefly cover.

  Ingesting HTTP and Syslog data

  After standing up Kafka and Kafka Connect, we wrote a service to ingest HTTP data and deployed Logstash to handle ingesting Syslog data. In the early days of development, we wrote an HTTP connector that was deployed in the Kafka Connect infrastructure, but we scrapped it because we decided Kafka Connect should only handle transporting data that is already at rest either within Kafka or in an external system. Events that are being emitted and are not yet at rest (e.g., most of the data coming in via HTTP and Syslog) should go straight to Kafka. This gives us the flexibility to shut down our Kafka Connect infrastructure in times of emergency or maintenance without risking the loss of data. This has come in handy a few times and generally simplifies the Kafka Connect deployment.

  User interface

  Before building our slick web UI, we built a backend service to tie together all the various in-house and open-source components of the whole platform. This administrative service wrapped the Kafka Connect REST API and exposed a much simpler API that performed validation and filled in default parameters. It can also perform various other activities such as list Kafka topics, sample Kafka topics and create tables in the Hive metastore when new topics are written to the data warehouse. The web UI is a React application built on top of this unified admin API. The web UI has wizards to help Data Scientists set up their data pipelines and walk them through the process.

  A very common example user flow would be something like this: A Data Scientist sends their log data via HTTP to a new Kafka topic, navigates to the web UI, follows the wizard to set up an S3 sink that pumps their logs into a table in the data warehouse and voila, their data flows into a new Hive table in the data warehouse in real time.

  Data transformation

  Data often exists in a format that’s not quite what you need. For trivial use cases we were able to leverage Kafka Connect’s transforms functionality to munge the data. For nontrivial transformations, particularly when we want to join the streaming data on data at rest in the data warehouse, we built a lightweight, self-service stream processing engine in Python. Kafka Streams would have been nice to use for this, as it is more powerful than our version, but Kafka Streams does not currently support Python. Because Python is the crowd favorite among Data Scientists, we wanted to provide a Python interface for them so that it could be self-service.

  The stream processing engine provides a runtime environment with automated monitoring and log collection. Data Scientists can write an arbitrary Python function that runs on each event, and deploy it using tools we provide. A lot of the deployment infrastructure necessary to build this stream-processing engine existed already (we have a very similar tool for standing up HTTP endpoints), so building it was straightforward.

  Data model

  All the data that flows through the platform is formatted for JSON and conforms to a data model that we enforce at all the ingestion points. The data model is an envelope-style format that looks like this:

                     {
                       “metadata”: {
                         “id”: “<unique id>”
                         “timestamp”: “<creation timestamp>”
                         “version”: “<user defined version>”
                       },
                       “payload”: {
                         ... arbitrary json blob ...
                       },
                       “event_name”: “<usually topic name, but not always>”
                     }

  This style of data formatting was a holdover from the legacy event system and is not necessarily be my ideal solution. To make the migration as seamless as possible, we decided to keep the JSON, envelope-style data model and enhance it with the addition of the metadata blob. Although having such a simple way to encode data has actually been pretty convenient, it is inefficient in terms of disk, memory and CPU usage. We’re limited in our ability to guarantee that events will have the fields Data Scientists expect them to have. It also doesn’t play nicely with Kafka’s ecosystem, which generally expects topics to have schemas. This has been a headache in some situations, particularly when installing new Kafka Connect connectors. If you are building a system like this from scratch, I heartily recommend that you start with schemas right off the bat so you never have to migrate to schemas later. Schemas, a schema registry and data lineage are all things we are just now planning to explore.

  Log compaction

  When data is flushed to the data warehouse, it is written in small batches of files. These small and numerous files clog up query systems such as Apache Spark and Presto, so we wrote a log compaction service that asynchronously compacts the small files into larger ones. This was actually a part of our legacy event infrastructure that we were able to continue using. It lets us write data to the data warehouse that can be queried in real time without cluttering the data warehouse with tiny files. Eventually, we may look into Hudi to handle this logic.

  Instrumentation and monitoring

  A crucial part of any system is the ability to know if it’s working or not. For this, we built a system of tracer bullets whereby we send special tracer events into every possible ingestion point, and route the data to every possible destination. When we send a tracer bullet, we persist a piece of metadata about it, including its unique ID, sent timestamp, ingestion point and the list of destinations where the tracer bullet should eventually arrive. We continuously query the metadata and the destination systems awaiting the arrival of the tracer bullets, measuring the total latency when they do arrive, and alerting on any lost or duplicate tracer bullets. This method of ensuring the integrity of the system is nice because it uses exactly the same interfaces that the users experience, and it is easily extensible. It does not depend on emitting metrics upon every write and read into the system; the caveat, however, is that you cannot always guarantee the integrity of the entire system at all times, because it is more like sampling the flow of data rathder than measuring all the data. For our use case this tradeoff was OK; however, we are investigating ways to make our monitoring more complete.

  Conclusion

  Never have I worked on a project this challenging that involved writing so little code. We spent about two-thirds of the project timeline on research, design, debate and prototyping. Over the months, we whittled away at every component until it was as small and maintainable as possible. We tried to salvage as much as we could from our legacy infrastructure. More than once we submitted patches upstream to Kafka Connect to avoid extra complexity on our end. We still strive to keep our Kafka infrastructure as current as possible so that we can continue to submit patches whenever necessary.

  The Data Highway has been running in production for almost eight months now, and we have yet to experience a major outage (knock on wood). The department has created several hundred topics with hundreds of connectors running in our distributed Kafka Connect deployment. The infrastructure hums along smoothly with very little operational burden or manual administration, freeing up our time to work on other impactful projects while giving Data Scientists autonomy and freedom.

  I recall a moment early on in the project when I overheard a conversation between a Data Scientist and engineer on one of the product teams. They were discussing ways that they could reliably pass data back and forth between their system, but their conversation fizzled out in frustration because at that time there was no good way to do so. Considering the hundreds of topics and connectors that our Data Scientists created in the past year, there must have been dozens of similarly frustrating conversations taking place all across the department. Now our Data Scientists have the freedom and autonomy to own their event-driven destinies, and it’s tremendously gratifying to know that those conversations are no longer ending in frustration but in fulfillment.

  The post Putting the Power of Apache Kafka into the Hands of Data Scientists appeared first on Confluent.

  
  
  https://www.confluent.io/blog/putting-power-apache-kafka-hands-data-scientists/
• 

  How to Build a UDF and/or UDAF in KSQL 5.0
  ksql confluent platform udaf udf • • sorangutan  

  1
  1
  Posts
  97
  Views

  

  KSQL is the open source streaming SQL engine that enables real-time data processing against Apache Kafka^®. KSQL makes it easy to read, write and process streaming data in real time, at scale, using SQL-like semantics. KSQL already has plenty of available functions like SUBSTRING, STRINGTOTIMESTAMP or COUNT. Even so, many users need additional functions to process their data streams.

  KSQL now has an official API for building your own functions. As of the release of Confluent Platform 5.0, KSQL supports creating user-defined scalar functions (UDF) and user-defined aggregate functions (UDAF). All it takes to write your own is just one Java class. The UD(A)F can embed any additional dependencies (like additional JARs or other binaries) if needed.

  In this blog post, we’ll discuss the main need and motivation for UD(A)Fs in KSQL, show an advanced example with a deep-learning UDF, prove how easy it is to build even something that is very powerful and identify some issues that you might face during development and testing.

  Motivation for KSQL UDFs/UDAFs

  Let’s first discuss what UD(A)Fs are and why you might need to build your own ones.

  Built-in KSQL functions to process streaming data

  Functions are one of the key components in KSQL. Functions in KSQL are like functions, operations or methods in other SQL engines or programming languages, such as Java, Python or Go. They accept parameters, perform an action—such as a complex calculation—and return the result of that action as a value. They are used within a KSQL query to filter, transform or aggregate data.

  Many functions are already built into KSQL, and more are added with every release. We distinguish them into two categories: stateless KSQL scalar functions and stateful KSQL aggregate functions.

  But what do you do if you miss a specific function in the KSQL syntax? Pretty simple: You build your own one with the official KSQL API. Let’s see how easy it is to write your own stateless UDF or stateful UDAF.

  UD(A)Fs in KSQL

  In contrast to built-in functions, UD(A)Fs are functions provided by the user of a program or environment. A UD(A)F provides a mechanism for extending the functionality of the KSQL engine by adding a function that can be evaluated in a standard query language statement. Using a UD(A)F in KSQL looks exactly like using built-in functions in KSQL. Both are registered to the KSQL engine before startup. They expect input parameters and return output values.

  Here is an example of the built-in function STRINGTOTIMESTAMP to convert a string value in the given format into the BIGINT value representing the timestamp:
  

  This function takes the timestamp 2017-12-18 11:12:13.111 (data type: String) and converts it to 1513591933111 (data type: BigInt).

  Now let’s take a look at the steps to build your own UDF to use it in your KSQL statements the same way as the STRINGTOTIMESTAMP function does above.

  Deep-learning UDFs for anomaly detection

  Building KSQL UDFs is easy, no matter if they are simple or powerful. With that in mind, let’s take a look at the steps to develop, test and deploy a new KSQL UDF.

  Steps to create, deploy and test a KSQL UDF

  You can build very simple KSQL functions like a specific scalar function for multiplication (see Confluent documentation to find out how), a really powerful function with many computations or aggregations, or anything in between these two extremities. Creating a stateful UDAF works the same way, of course.

  The example below shows you how to build a more advanced UDF, which embeds an analytic model (i.e., an external dependency). This sounds complex, but in reality the implementation is simple. We write one Java class, add the dependency, write our business logic and build an uber JAR which includes the UDF and its dependency. This UDF can be deployed to the KSQL server(s) and then used in KSQL statements by end users without any coding in a programming language.

  Java source code of the deep-learning UDF

  Develop a UDF is quite easy, too. Just implement the function in a Java method within a Java class and add the corresponding Java annotations:
  

  Here is the full source code for the anomaly detection KSQL UDF:
  

  As you can see, you can really focus on implementing the business logic. In my example, I consume new events and do stateless processing by applying an analytic model. The prediction is returned immediately. Under the hood, KSQL receives events from a Kafka topic and sends the output to another Kafka topic.

  The wrapper code for the UDF is minimal. You only need to add two annotations and the corresponding import statements to your POJO (plain old Java object):

  • Import io.confluent.ksql.function.udf.Udf and io.confluent.ksql.function.udf.Udf
  • @UDFDescription: class annotation to describe the name and functionality of the UDF
  • @UDF: method annotation to describe the functionality of the specific method (there can be more than one of these methods in a UDF class to process different interface signatures)

  And that’s it. The annotations in this class make sure the KSQL server interprets this class correctly and loads the UDF into the runtime engine during startup.

  Deployment to KSQL server(s)

  The deployment of the UDF is pretty straightforward, no matter if you’re using a local installation on your laptop or a cluster of remote KSQL servers:

  • Build an uber JAR that includes the KSQL UDF and any dependencies. Typically, you use a build tool like Apache Maven or Gradle to do this. See the pom.xml of my Github project for an example using Maven.
  • Copy the uber JAR to the ext/ directory that is part of the KSQL distribution. The ext/ directory can be configured via the property ksql.extension.dir in KSQL configuration file ksql-server.properties. If this directory does not exist yet, simply create a new directory.
  • Start/restart your KSQL server instance(s) to pick up new UD(A)Fs.
  • Try out the new UDF in KSQL command line interface via Confluent REST Proxy or Confluent Control Center’s KSQL user interface, which has nice features like autocompletion.
  • If the UDF does not work, check the log’s errors. They will appear in the KSQL server log (ksql.log).

  Working demo: MQTT sensor analytics with Apache Kafka and KSQL

  Let’s now think about a practical example for a powerful UDF: continuous processing of millions of events from connected devices, specifically in the case of car sensors.

  This use case leverages a UDF that does real-time predictions using an autoencoder neural network to detect anomalies. You can apply the analytic model by executing the KSQL statement in which “anomaly” is the UDF:
  

  You can also create a new KSQL stream to deploy the statement for continuous processing. The following example filters values over a specific threshold of five to send alerts if an anomaly occurs:
  

  This KSQL stream AnomalyDetectionWithFilter scales like any other KSQL or Kafka Streams application. It is built for S, M, L, X and XXL scenarios. If you need higher scalability or better throughput, simply create additional instances on KSQL servers.

  If you want to try out this machine-learning UDF, just go to my Github project. It includes the source code and a step-by-step guide to test it using Confluent CLI and Confluent MQTT Proxy, which includes a MQTT sensor generator.

  The story around building and deploying analytic models with KSQL is a very interesting and compelling use case worth discussing in further detail, but I’ll save that discussion for a future blog post that covers model training, deployment and inference in greater depth.

  Potential issues during development and testing

  During development and testing, I faced some challenges when creating my first UDF with the new KSQL REST API. If your UDF does not work, check for the following:

  • Compile errors? Did you include KSQL JAR into your project to use the classes and annotations? Either add ksql-udf.jar, which you get from a KSQL project build, or add it via Maven dependency (groupId: io.confluent.ksq; artifactId: ksql-udf).
  • Classpath errors at runtime? Are dependencies missing in the generated uber JAR? Extract the JAR and see if all needed dependencies are included (e.g., the analytic model in my example) and in the expected directory. Maybe your build process is not working well yet.
  • UDF not found at runtime? Do you see the error, “Can’t find any functions with the name XYZ”? Did you restart the KSQL server after you added the uber JAR to the /ext directory? UDFs are only loaded during the startup of the KSQL engine. Or, perhaps you have the wrong path? Did you configure the right path in property ksql.extension.dir in ksql-server.properties? Maybe start with an absolute path first. If that works, you can try a relative path.
  • Still experiencing problems? Check the logs—not just the KSQL log (ksql.log) but also other Kafka-related log files.

  Please let us know if you face any other issues during development and testing of KSQL UDFs. All feedback is appreciated. While this blog post focused on building a stateless UDF, the development process is the same for a stateful UDAF.

  What’s next for KSQL UD(A)Fs?

  The foundation is here. You’ve learned how easy it is to create and deploy a new UD(A)F for KSQL, involving only a few steps to implement, register and execute a KSQL UDF. Once the developer builds it, then every end user of your KSQL instances can leverage it whether they’re a data engineer or data scientist.

  The example of building a deep-learning UDF shows how you can even include other dependencies, such as an analytic model, to build very powerful functions. The end user just has to call the function in the KSQL statement—no need to understand what is going on under the hood.

  The next step in the KSQL UD(A)F roadmap is to support building UD(A)Fs with programming languages other than Java. This way, even a data scientist who only understands Python or JavaScript can write and deploy their own functions. In addition to this, there are many new and exciting features to come, so be sure to keep an eye out for the next KSQL releases.

  If you’re interested in what KSQL can do:

  • Watch the introductory video of KSQL
  • Download Confluent Platform and follow the quick start to begin using KSQL
  • Check out the demo on building a stream application with KSQL
  • Ask questions in the #ksql channel of our community Slack group

   

  The post How to Build a UDF and/or UDAF in KSQL 5.0 appeared first on Confluent.

  
  
  https://www.confluent.io/blog/build-udf-udaf-ksql-5-0
• 

  Kafka Streams in Action
  ksql stream processing kafka streams kafka streams api • • sorangutan  

  1
  1
  Posts
  108
  Views

  

  Confluent’s own Bill Bejeck has recently completed Kafka Streams In Action, a book about building real-time applications and microservices with the Kafka Streams API. The book is scheduled to be available next month, but Manning Publications has kindly agreed to let us share my foreword to the book here on the Confluent blog. If you’re interested in learning more about Kafka Streams (and you should be!), this book is an excellent way to get started.

  Foreword

  We are witnessing the emergence of a new paradigm for building modern data-centric applications—event-driven architectures powered by Apache Kafka^®. The traditional approach to building tightly coupled request/response applications around data transactions is being rapidly augmented by event-driven applications that are instead centered around business events, leading to a far more responsive, flexible, decoupled application architecture. We can now see event-driven applications at the heart of next-generation architectures in every industry imaginable. Big retailers are re-working their fundamental business processes around continuous event streams; car companies are collecting and processing real-time event streams from internet-connected cars; and banks are rethinking their fundamental processes and systems around Kafka as well.

  What’s powering this paradigm shift? If the world has lived with request/response applications and batch computing for so long, why is this happening now?

  First, businesses are far more digital than they used to be. And this digitization is quickly going beyond the domain of traditional enterprise software systems. The “Internet of Things” refers to many changes, but more than anything, it refers the process of automatically instrumenting things happening in the physical world—from manufacturing processes to automobiles to retail inventory; aspects of businesses that were once off limits to computers are becoming sources of continuous event streams.

  Second, there is also a transformation in how data is used inside organizations. In a world where action was only taken by humans, periodic batch-computed reporting might be plenty fast. After all, no human will check the report more than once a day. But in a company which has become significantly digital, in which actions are often taken not only by humans but automatically by software, this kind of delay makes no sense. This shift towards modelling processing around event streams actually makes things simpler. The core data of most modern businesses is most naturally thought of as a continuous event stream (of sales, customer experiences, shipments, etc.). It is natural that the processing and analysis of this data—via event-driven applications—would be continuous, and real-time as well.

  The final driver of the change is that the technology for supporting stream processing has gotten much better. Early systems, including things like Enterprise Messaging systems, and Complex Event Processing engines, existed in this space, but were quite limited in the types of problems they could handle. As part of building Apache Kafka, we have learned a great deal about how to create stream processing systems so that they can not only scale horizontally to company-wide usage, but also so that their capabilities are truly a superset rather than a subset of what was possible with batch processing. This area is still evolving rapidly, but forward-thinking companies are already implementing production stream processing applications in critical domains. The rise of Apache Kafka as an Event Streaming Platform that powers stream processing is evidence to this trend of stream processing going mainstream.

  So what is stream processing? And what role does it play in enabling an event-driven architecture?

  Stream Processing is a data processing paradigm that views data less as something kept in static stores and more as something that flows and is processed continuously. Stream Processing manifests itself in event-driven applications that power the business, as well as, in real-time analytics that report on the business.

  Stream Processing requires a fundamental shift in our thinking, away from command thinking towards event thinking; a cultural change that enables responsive, event-driven, extensible, flexible, and real-time applications. In business, event thinking opens organizations to real-time, context-sensitive decision making and operations. In technology, event thinking can produce more autonomous and decoupled software applications and, as a consequence, elastically scalable and extensible systems. In both cases, the ultimate benefit is greater agility—of the business and of the business-facilitating technology. Applying event thinking to an entire organization is the foundation of the event-driven architecture. And stream processing is the technology that enables this transformation.

  So what is Kafka Streams? Why did we build it? And how is it different from all the other Big Data processing frameworks?

  Kafka Streams is the stream processing library native to Apache Kafka for building event-driven applications in Java to process data in Apache Kafka topics. Applications that use this library can do sophisticated transformations on data streams that are automatically made fault-tolerant and are transparently and elastically distributed over the instances of the application. Since its initial release in the 0.10 version of Apache Kafka back in 2016, many companies have put Kafka Streams into production, including Pinterest, New York Times, Rabobank, LINE, and many more.

  There are a wide variety of technologies, frameworks, and libraries for building applications that process streams of data. Big Data processing frameworks such as Storm, Flink, and Spark all have their pros and cons. Kafka’s approach to stream processing—be it Kafka Streams or KSQL—is pretty different from these big data frameworks. Our goal with Kafka Streams and KSQL is to make stream processing simple enough that it can be a natural way of building event-driven applications that respond to events, not just a heavy-weight “big data” thing.

  It’s because we tried building a stream processing framework by modeling it as a faster MapReduce layer, and failed. Apache Samza was our initial take on stream processing with Apache Kafka. Well, technically our second take, as the first one was a prototype I called KafkaMR. But adoption of Apache Samza, even inside the company it was introduced at, didn’t quite pan out as expected. The reason is that this big data approach to stream processing poses it as a problem of multiplexing a set of jobs over a cluster of machines. While stream processing is most naturally expressed as event-driven applications where business logic of an application embeds the ability to process events streams.

  Kafka’s approach to stream processing—be it Kafka’s Streams API or KSQL—is hence pretty different from these big data frameworks. Rather than model stream processing applications as a big data framework, Kafka models them after REST applications. Rather than conceive stream processing as a kind of Java big data processing framework that happens to transmit streams of data, we invert this. In our model, the primary entity isn’t the processing code at all, it’s the streams of data in Kafka.

  Kafka provides a versioned, widely adopted protocol for correct, distributed, fault-tolerant, stateful stream processing. This protocol can be used from any language. One example of this is the Kafka Streams API in Apache Kafka, another is KSQL, and there are several more in Go, .Net, Python, Node, etc. Each of these isn’t all that much code, and the code is, in any case, essentially an implementation detail. As with a REST service, the contract is the input and output of the service, not the thing you used in the internals of your app. And that is the reason Kafka’s approach to stream processing is a powerful enabler of modern stream processing.

  I believe that the event-driven architecture, centered around real-time event streams and stream processing, will become ubiquitous in the years ahead. Technically sophisticated companies like Netflix, Uber, Goldman Sachs, Bloomberg and others have built out this type of large-scale event streaming platform, many operating at massive scale. It’s a bold claim, but I think the emergence of stream processing and the event-driven architecture will have as big an impact in reworking how companies make use of data as relational databases did.

  Still, learning event thinking and building event-driven applications oriented around stream processing is quite a mindshift if you are coming from the world of request/response style applications and relational databases. This book is a great way to learn about Kafka Streams, and learn how it is a key enabler of event-driven applications. I hope you enjoy reading it as much as I have!

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

  • Watch our introductory video of KSQL
  • Download Confluent Platform and follow the quick start to begin using KSQL
  • 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
  • Play with our Confluent demo applications

  The post Kafka Streams in Action appeared first on Confluent.

  
  
  https://www.confluent.io/blog/kafka-streams-action
• 

  Getting Started with Apache Kafka and Kubernetes
  kubernetes apache kafka confluent platform • • sorangutan  

  1
  1
  Posts
  86
  Views

  

  Enabling everyone to run Apache Kafka^® on Kubernetes is an important part of our mission to put a streaming platform at the heart of every company. This is why we look forward to releasing an implementation of the Kubernetes Operator API for automated provisioning, management, and operations of Kafka on Kubernetes. (Read about that here and let us know if you’d like to take it for a spin when it’s available!)

  Today, we are taking a step in this direction by sharing two resources to help you run Confluent Platform on Kubernetes:

  1. Helm Charts that provide a starting point for deployment of Confluent Platform components
  2. A white paper that shares the best practices we’ve learned by running Confluent Platform on Kubernetes

  Getting started with deployment templates

  Now available on GitHub in developer preview are open-source Helm Chart deployment templates for Confluent Platform components. These templates enable developers to quickly provision Apache Kafka, Apache ZooKeeper, Confluent Schema Registry, Confluent REST Proxy, and Kafka Connect on Kubernetes, using official Confluent Platform Docker images.

  Helm is an open-source packaging tool that helps you install applications and services on Kubernetes. Helm uses a packaging format called charts. A chart is a collection of YAML templates that describe a related set of Kubernetes resources.

  For stateful components like Kafka and ZooKeeper, the Helm Charts use both StatefulSets to provide an identity to each pod in the form of an ordinal index, and Persistent Volumes that are always mounted for the pod. For stateless components, like REST Proxy, the Helm Charts utilize Deployments instead to provide an identity to each pod. Each component’s charts utilize Services to provide access to each pod.

  It’s really easy to get started. Just install Helm on your local or deployed Kubernetes cluster, and use the parent chart to configure and deploy Confluent Platform components:

  helm install cp-helm-charts

  

  Helm allows you to customize configurations using YAML templates. For example, to deploy five Kafka brokers and set a custom `min.insync.replicas`, create a copy of the default YAML template, update the values, and use it to install the configuration:

  helm install -f custom-values.yaml cp-helm-charts
  
  custom-values.yaml
  
  … 
  
  ## Number of Kafka brokers 
  
  brokers: 5
  
  …
  
  ## Kafka Server properties
  
  configurationOverrides:
  
    "log.dirs": /opt/kafka/data/logs
  
    "offsets.topic.replication.factor": 3
  
    "min.insync.replicas": 3
  
  … 

  Special thanks to Qi Shao, Nikhil Chandrappa, Li Wang, Amey Banarse and Prasad Radhakrishnan from Pivotal’s (NYSE: PVTL) Platform Architecture team for their contributions in developing and testing these Helm Charts.

  Sharing best practices

  What’s both exciting and challenging about working with the vibrant Kubernetes ecosystem is that everything progresses so fast. To help you understand and plan for running Confluent Platform on Kubernetes, we’ve provided a white paper that shares the best practices we’ve learned over time.

  The Recommendations for Deploying Apache Kafka on Kubernetes white paper helps you see how Confluent Platform components fit into the Kubernetes ecosystem. It also covers our approach to networking, storage, traffic, log aggregation, metrics and more.

  We will update this white paper as the Kubernetes ecosystem continues to evolve.

  Conclusion

  Our end goal for this is to make streaming data ubiquitous. Kubernetes lets you run your applications and services anywhere. Kafka enables you to make your data accessible instantaneously, anywhere.

  Get started with Kafka on Kubernetes today by checking out the white paper and Helm Charts on our website.

  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’ll notify you when it’s available!

  The post Getting Started with Apache Kafka and Kubernetes appeared first on Confluent.

  
  
  https://www.confluent.io/blog/getting-started-apache-kafka-kubernetes/
• 

  Decoupling Systems with Apache Kafka, Schema Registry and Avro
  confluent platform avro schema registry • • sorangutan  

  1
  1
  Posts
  102
  Views

  

  As your Apache Kafka^® deployment starts to grow, the benefits of using a schema registry quickly become compelling. Confluent Schema Registry, which is included in Confluent Open Source, enables you to achieve strong decoupling of the systems you integrate via Kafka, in turn allowing your teams to be more agile and create applications that are more robust to change.

  Why you want a schema registry

  Kafka producers and consumers are already decoupled in the sense that they do not communicate with one another directly; instead, information transfer happens via Kafka topics. But they are coupled in the sense that consumers need to know what the data they are reading represents in order to make sense of it—but this is something that is controlled by the producer! In the same way that Kafka acts as an intermediary for your data, Schema Registry can act as an intermediary for the types of data that you publish to Kafka.

  In any non-trivial system, requirements change over time and the schemas needed to satisfy them change accordingly; that is, schemas evolve. This evolution makes this form of coupling even stronger. Not only do consumers need to understand the type of data they are reading today, they also depend on any changes that might be made to this type over time. You can reduce this coupling by defining constraints on the way schemas are allowed to evolve upfront and manage these constraints using Schema Registry as well. Producers agree to adhere to the published evolution rules, and consumers are written to be able to handle any schema that is consistent with them. The downside of this, of course, is that it limits flexibility and adds some overhead due to additional formality. However, the payoff in terms of reduced coupling is usually much greater.

  Confluent Schema Registry is built around the Apache Avro serialization format. One of the reasons Avro was selected as the recommended format for use with Confluent Platform is that it has flexible, well-defined rules around schema evolution. Jay Kreps expands on this and other appealing aspects of Avro in his article Why Avro for Kafka Data?

  	Side Note: Avro is often compared with Protobuf and Thrift, two other commonly used serialization formats. A key way in which Avro differs from these formats is that it requires that the schema used to serialize data be available when parsing it. Protobuf and Thrift do not—the serialized data includes enough meta-information to allow it to be parsed without a schema. But this meta-information is only a partial description of the data. For example, it doesn’t include field names or doc strings, so it doesn’t replace the role of a schema registry.

  Putting it into practice with .NET

  In this blog post, we’ll walk through a practical example of how to use Avro and Schema Registry from C#/.NET. We’ll assume that you have a Kafka cluster and Schema Registry up and running already (for more information, refer to the Confluent Platform Quick Start). We’ll also assume that you’re set up to build and run C# applications, either targeting the .NET Framework or .NET Core on your platform of choice.

  The .NET Avro serializer and deserializer allow you to work with Avro data in one of two ways:

  1. Via specific classes generated using the avrogen tool
  2. Using the GenericRecord class to read and write records with any schema, without needing to pre-generate any class files

  We’ll demonstrate how to use both, starting with GenericRecord, then the specific compiler-generated classes. In general, you’ll find working with the specific classes much simpler, and you should prefer this where possible. Use GenericRecord in scenarios where you need to work dynamically with data of any type.

  First, let’s define a schema for a simple log message type. We’ll specify this over two files: one for the LogMessage itself, and the other for the LogLevel enumeration that it depends on:

  {
    "namespace": "MessageTypes",
    "type": "enum", 
    "doc": "Enumerates the set of allowable log levels.",
    "name": "LogLevel", 
    "symbols": ["None", "Verbose", "Info", "Warning", "Error"]
  }
  
  { 
    "namespace": "MessageTypes",
    "type": "record",
    "doc": "A simple log message type as used by this blog post.",
    "name": "LogMessage",
    "fields": [
      { "name": "IP", "type": "string" },
      { "name": "Message", "type": "string" },
      { "name": "Severity", "type": MessageTypes.LogLevel }
    ]
  }

  The Avro format allows you to specify documentation along with your schemas using the doc attribute. You should make good use of this feature. Schemas should be self-contained—someone using your schema should not need to consult any other source of information. Pro tip: you can auto-generate user-readable documentation for your schemas using the Avrodoc tool.

  Below is an example of how to produce a LogMessage message to Kafka conforming to the above schema. (Note: The complete source code for all examples is available on GitHub.)

  var config = new Dictionary<string, object>
  {
      { "bootstrap.servers", bootstrapServers },
      { "schema.registry.url", schemaRegistryUrl }
  };
  
  using (var producer = new Producer<Null, GenericRecord>(
    config, null, new AvroSerializer<GenericRecord>()))
  {
      var logLevelSchema = (EnumSchema)Schema.Parse(
          File.ReadAllText("LogLevel.asvc"));
  
      var logMessageSchema = (RecordSchema)Schema
          .Parse(File.ReadAllText("LogMessage.V1.asvc")
              .Replace(
                  "MessageTypes.LogLevel", 
                  File.ReadAllText("LogLevel.asvc")));
  
      var record = new GenericRecord(logMessageSchema);
      record.Add("IP", "127.0.0.1");
      record.Add("Message", "a test log message");
      record.Add("Severity", new GenericEnum(logLevelSchema, "Error"));
      producer.ProduceAsync("log-messages", null, record)
          .ContinueWith(dr => Console.WriteLine(dr.Result.Error 
              ? $"error producing message: {dr.Result.Error.Reason}"
              : $"produced to: {dr.Result.TopicPartitionOffset}"));
  
      producer.Flush(TimeSpan.FromSeconds(30));
  }

  There are a few things worth noting:

  • In addition to the required producer property bootstrap.servers, you’ll need to specify a value for schema.registry.url, which is used by the Avro serializer to communicate with Schema Registry.
  • The Avro specification does not provide a way for one schema file to reference another. However, in order to avoid duplicating the LogLevel schema definition, we need to do this because we’re required to specify it separately when constructing the GenericEnum Severity value. Instead of relying on Avro, we use string substitution in our C# program. The LogMessage schema specified above is not a valid Avro schema prior to doing this substitution.
  • By default, the serializer will automatically register new schemas with Schema Registry where compatibility rules allow. Since we haven’t yet registered any other schema against this topic, our LogMessage schema will be automatically registered before the message is produced.
  • Behind the scenes, each schema registered with Schema Registry is given a unique ID that is prepended to the serialized data in each message sent to Kafka. When a consumer reads a message, it uses this ID to associate the message with the schema that was used to write it. If the consumer encounters data that was serialized with a schema it has not seen yet, it will contact Schema Registry to retrieve the schema.

  You can verify that the schema was added to Schema Registry as follows:

  curl http://localhost:8081/subjects
  
  > ["log-messages-value"]

  Notice that the schema was registered under the name log-messages-value, which is the topic name concatenated with -value. This name is referred to as a subject. Corresponding to any given topic, there are potentially two subjects registered in Schema Registry—one corresponding to the message value and one corresponding to the message key (i.e., the topic name appended with -key).

  You can interact with Schema Registry further using its REST API. Something else you might want to do is set the subject compatibility to FULL (by default, it’s BACKWARD):

  curl -X PUT -H "Content-Type: application/vnd.schemaregistry.v1+json" \
      --data '{"compatibility": "FULL"}' \
      http://localhost:8081/config/log-messages-value

  Full compatibility enforces that any schema written to this subject is both forwards and backwards compatible with older schemas:

  • A schema is backward compatible if it can be used to read data written with a previous version of the schema.
  • A schema is forward compatible if previous versions of the schema can be used to read data written by it.

  From the perspective of decoupling producers and consumers, forwards compatibility is very desirable because it ensures consumers will be able to read new versions of data without being updated. Backwards compatibility is important for data destined for archival storage because it ensures that the latest schema can be used to read all historical data.

  It’s common to want to evolve a schema by adding another field. Conveniently, the new schema will be both forwards and backwards compatible with the old one, provided you specify a default value for the added fields. For a full set of forwards/backwards compatibility rules, refer to the documentation on data serialization and evolution.

  OK, let’s add a tags field to our original schema, and this time we’ll define the LogLevel enum inline:

  { 
    "namespace": "MessageTypes",
    "type": "record",
    "doc": "A simple log message type as used by this blog post.",
    "name": "LogMessage",
    "fields": [
      { "name": "IP", "type": "string" },
      { "name": "Message", "type": "string" },
      { "name": "Tags",
        "type": { "type": "map",
                  "values": "string"},
                  "default": {}},
      { "name": "Severity",
        "type": { "namespace": "MessageTypes",
                  "type": "enum",
                  "doc": "Enumerates the set of allowable log levels.",
                  "name": "LogLevel",
                  "symbols": ["None", "Verbose", "Info", "Warning", "Error"]}}
    ]
  }

  Instead of making use of the GenericRecord class, let’s now use the avrogen tool to generate a C# class corresponding to the new schema. avrogen is available as a .NET Core Global Tool (requires .NET Core 2.1 or above). It can be installed as follows:

  dotnet tool install -g Confluent.Apache.Avro.AvroGen

  Now, you can run the tool in your project directory:

  avrogen -s LogMessage.V2.asvc .

  And then use the generated class directly with AvroSerializer:

  var config = new Dictionary<string, object>
  {
      { "bootstrap.servers", bootstrapServers },
      { "schema.registry.url", schemaRegistryUrl }
  };
  
  using (var producer = new Producer<Null, MessageTypes.LogMessage>(
    config, null, new AvroSerializer<MessageTypes.LogMessage>()))
  {
      producer.ProduceAsync(
          "log-messages", null, 
          new MessageTypes.LogMessage
          {
              IP = "192.168.0.1",
              Message = "a test message 2",
              Severity = MessageTypes.LogLevel.Info,
              Tags = new Dictionary<string, string> { { "location", "CA" } }
          });
      producer.Flush(TimeSpan.FromSeconds(30));
  }

  Or with AvroDeserializer:

  var consumerConfig = new Dictionary<string, object>
  {
      { "group.id", Guid.NewGuid().ToString() },
      { "bootstrap.servers", bootstrapServers },
      { "schema.registry.url", schemaRegistryUrl },
      { "auto.offset.reset", "beginning" }
  };
  
  using (var consumer = new Consumer<Null, MessageTypes.LogMessage>(
     consumerConfig, null, new AvroDeserializer<MessageTypes.LogMessage>()))
  {
      consumer.OnConsumeError 
          += (_, error) => Console.WriteLine($"an error occured: {error.Error.Reason}");
  
      consumer.OnMessage 
          += (_, msg) => Console.WriteLine($"{msg.Timestamp.UtcDateTime.ToString("yyyy-MM-dd HH:mm:ss")}: [{msg.Value.Severity}] {msg.Value.Message}");
  
      consumer.Subscribe("log-messages");
  
      while (true)
      {
          consumer.Poll(TimeSpan.FromSeconds(1));
      }
  }

  When used in the context of the consumer, the MessageTypes.LogMessage schema is referred to as the reader schema. Messages read by the consumer may contain data that was serialized using this schema or potentially any other schema (referred to as the writer schema). In the event that the reader and writer schemas are different, Avro schema resolution rules will be used to reconcile the difference. If the schemas are incompatible, the consumed message will be delivered via the OnConsumeError event rather than the OnMessage event.

  With the above examples, we’ve shown how straightforward it is to use Confluent Schema Registry and Avro serialized data with your .NET applications. People just starting out with Kafka often choose to work with a simple text-based serialization format such as JSON, but as your Kafka deployment starts to grow in complexity, there is a lot to be gained from using a data format that provides strong schema evolution rules and by employing a central schema registry to manage them.

  If you enjoyed this article on Confluent Schema Registry and Avro, you might also enjoy:

  • How I Learned to Stop Worrying and Love the Schema by Geoff Anderson
  • Yes, Virginia, You Really Do Need a Schema Registry  by Gwen Shapira
  • Why Avro for Kafka Data? by Jay Kreps for an in-depth discussion of the Avro serialization format
  • Confluent’s .NET Client for Apache Kafka GitHub repository to access the complete source code for all examples in this blog post
  • Documentation on Schema Registry, where you’ll find a working demo of Schema Registry being used for stream processing
  • The Confluent website where you can download Confluent Platform
  • The Confluent Docker image for Schema Registry

  The post Decoupling Systems with Apache Kafka, Schema Registry and Avro appeared first on Confluent.

  
  
  https://www.confluent.io/blog/decoupling-systems-with-apache-kafka-schema-registry-and-avro/
• 

  Welcoming the Distributed Masonry Team to Confluent
  company confluent cloud distributed masonry • • sorangutan  

  1
  1
  Posts
  84
  Views

  

  Streaming platforms force a fundamental rethink of the way companies treat the data that describes and drives their business, taking us away from the static notion of data which sits still, in a database, to a more dynamic one where events move around the organization, triggering applications into action, materializing views in different places or being manipulated directly as they flow. This is a vision we’ve believed in for some time, but it’s refreshing to see independent parties come to similar conclusions, and even more so when when they put those conclusions into action.

  With this in mind, we are very pleased to announce that the Seattle-based Distributed Masonry team, the innovators behind Onyx Platform and Pyrostore, will be joining Confluent to extend the cloud-native storage and processing capabilities of Confluent Cloud. Not only does the Distributed Masonry team share a similar vision of the future, they made their interpretation a reality with a product whose features reach far beyond a simple ‘lift and shift’ of Kafka onto the cloud.

  Innovations like these are important because cloud-native technologies are shaped by a different set of forces to those built for on premise infrastructure, and while streaming platforms are used broadly across every major market segment, we have merely scratched the surface of what is to come. Stay tuned to find out how the Distributed Masonry team and Confluent will help shape this future together.

  The post Welcoming the Distributed Masonry Team to Confluent appeared first on Confluent.

  
  
  https://www.confluent.io/blog/welcoming-the-distributed-masonry-team-to-confluent/
• 

  Introducing Confluent Platform 5.0
  ksql confluent platform confluent control center replicator schema registry mqtt • • sorangutan  

  1
  1
  Posts
  75
  Views

  

  We are excited to announce the release of Confluent Platform 5.0, the enterprise streaming platform built on Apache Kafka^®. At Confluent, our vision is to place a streaming platform at the heart of every modern enterprise, helping infrastructure owners get the most out of Kafka and empowering developers to build powerful applications with real-time, streaming data. With a comprehensive streaming platform, companies can better integrate all their disparate data sources into a single source of truth and respond in real time to every event affecting the business.

  Since the release of Confluent Platform 1.0, we have provided an enterprise-class streaming platform that includes the latest in streaming technology based on Apache Kafka. Our latest release introduces significant new capabilities, from making infrastructure more secure, reliable and easier to manage, to enabling more powerful applications with streaming data.

  With Confluent Platform 5.0, operators can secure infrastructure using the new, easy-to-use LDAP authorizer plugin and can deliver faster disaster recovery (DR) thanks to automatic offset translation in Confluent Replicator. In Confluent Control Center, operators can now view broker configurations and inspect consumer lag to ensure that they are getting the most out of Kafka and that applications are performing as expected.

  We have also introduced advanced capabilities for developers. In Confluent Control Center, developers can now better understand the data in Kafka topics due to the new topic inspection feature and Confluent Schema Registry integration. Control Center presents a new graphical user interface (GUI) for writing KSQL, making stream processing more effortless and intuitive as well. The latest version of KSQL itself introduces exciting additions, such as support for nested data, user-defined functions (UDFs), new types of joins and an enhanced REST API. Furthermore, Confluent Platform 5.0 includes the new Confluent MQTT Proxy for easier Internet of Things (IoT) integration with Kafka. The latest release is built on Apache Kafka 2.0, which features several new functionalities and performance improvements.

  Download Confluent Platform 5.0

   

  Operate more secure, reliable and performant Apache Kafka

  Operationalizing enterprise infrastructure is hard, and Apache Kafka is no exception. Our mission at Confluent is to make managing Kafka easy, and the latest release of Confluent Platform helps operators do just that.

  Secure Kafka using the new LDAP authorizer plugin and control feature access in Confluent Control Center

  As customers adopt Confluent Platform for their mission-critical data, security becomes an important consideration. Operators need to be able to easily manage user access to sensitive data, but doing so in Kafka has historically been difficult. Because security is top-of-mind for many of our customers, we have made several investments to improve security management in Confluent Platform and will continue providing new security features in later releases.

  The majority of enterprises standardize on AD/LDAP for identity-related services. The latest release of Confluent Platform introduces an LDAP authorizer plugin to set up access control for Kafka, allowing customers to use the user-group mappings in their LDAP server. The plugin works with any security protocol by mapping an authenticated user principal to one or more groups and allowing user-principal-based ACLs, as well as group-based ACLs. This simplifies access control management because a smaller number of rules can be used to manage access to groups or organizations.

  Enterprises also 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 have included feature access controls for topic inspection, schemas, and KSQL. When customers restrict access to a feature via the configuration file, Control Center’s UI will reflect this change upon startup, and users cannot circumvent these protections in any way.

  Minimize disaster recovery time with automatic offset translation in Confluent Replicator

  The disaster recovery (DR) response process is complex. Operators need to ensure that the data is replicated, applications failover, and applications then switch back when the original data center (DC) has recovered. Replicator supports operators along each step of the process by providing automation that minimizes the work needed during a DR response.

  Replicator already automates the replication of topic messages and related metadata. With Confluent Platform 5.0,  it also ensures that consumer offsets are available on the secondary cluster. When consumer client applications failover to the secondary cluster, Replicator handles consumer offset translation so that they can resume consuming near the last point they stopped consuming at in the original cluster. This minimizes the reprocessing that consumers needs to do in a disaster scenario without skipping messages.

  To facilitate application failover and switch back in Confluent Platform 5.0, Replicator adds support for protection against circular replication. This guarantees that if two Replicator instances are configured to run, one replicating from DC1 to DC2, and the second instance configured to replicate from DC2 to DC1, Replicator will ensure that messages replicated to DC2 are not replicated back to DC1, and vice versa. As a result, Replicator safely runs in each direction.

  Monitor more and manage better using Confluent Control Center

  We built Control Center to deliver understanding and insight into the inner workings of Apache Kafka and streaming data, helping operators monitor and manage their Kafka installations as well as possible. Confluent Platform 5.0 introduces a broker configuration view that ultimately improves this experience. It empowers operators to see broker configurations across multiple Kafka clusters, check configuration values for specific brokers, compare differences within a cluster to identify potential risks such as mismatched security configurations in different brokers, and even download configurations. This new feature helps operators ensure that their brokers are set up correctly, enabling them to get the most out of Kafka.

  

  Once you have configured Kafka the way you want and have applications consuming data from topics, it’s important to understand how those applications are performing. Monitoring application performance is something that both operators and developers care about. Since its genesis, Control Center has included an advanced stream monitoring feature, which provides insight into end-to-end performance of the produce-consume cycle. In the latest release, we introduce consumer lag, which allows users to view how consumers are performing based on offset, spot potential issues at a glance, and take proactive steps to keep performance high. Users can also set alerts to monitor lag behind-the-scenes and notify them if issues arise.

  

  Build more powerful streaming applications with Kafka

  Having a secure, reliable, and performant Kafka infrastructure is the first part of being successful with streaming data, but it’s not enough. Along with helping enterprises better monitor and manage Kafka, we strive to make building applications with streaming data straightforward, seamless and fast. Confluent Platform 5.0 delivers new capabilities for application developers in both Control Center and KSQL.

  Understand and process streaming data with Confluent Control Center

  Control Center now allows users to gain insight into the actual data in Kafka topics via a new topic inspection feature. Users can see the streaming messages in topics and read key, header, and value data for each message. Thanks to integration with Schema Registry, topic inspection supports JSON, string, and Avro data formats.

  

  In addition to viewing message data, developers also need to understand how data is organized in order to build applications with Kafka. This need to understand data structure and manage compatibility quickly grows in importance as an enterprise scales up with Kafka. Confluent Schema Registry is a valuable tool for coordinating teams of developers, and makes managing a real-time streaming platform with multiple connected applications seamless. With the latest release of Confluent Platform, Control Center integrates with Schema Registry and allows users to see the key and value schemas for topics. Users can also view older schema versions and compare them against current versions in a git-like UI, which makes finding differences in evolving schemas and building applications on streaming data simple.

  

  Once developers can view message data and understand how data is structured, they have the basic building blocks to start building applications. Last year, we announced KSQL, the streaming SQL engine for Apache Kafka, to help developers build powerful applications with streaming data. While Confluent Platform 4.1 presented an experimental web interface for developers to run KSQL, in the latest release, Control Center includes KSQL as part of a new development area. With this new GUI for KSQL, stream processing has never been more accessible. Developers can create streams and tables from topics, experiment with transient queries, and run persistent queries to filter and enrich data. What’s more, they can do all this with the power of autocompletion, which lowers the bar to getting started with stream processing. No more trying to figure out what streams and tables exist, or what syntax is right for a command—autocomplete guides developers along the way. KSQL in Control Center lowers the bar for both developers building streaming applications as well as operators who wish to debug cluster issues by filtering and consuming data from topics.

  

  
  Build more powerful applications with KSQL

  Along with providing better GUI support for building applications in Control Center, the latest release includes several highly requested KSQL functionalities.

  The most commonly requested feature for KSQL has been support for nested data. Today, KSQL includes a STRUCT type that allows developers to work with nested data in both Avro and JSON formats. This means that you can use KSQL to process data from a broader variety of sources, and tasks such as data extraction and data conversion are even easier.

  Another highly requested feature has been the ability to define your own computations that go beyond what KSQL provides out-of-the-box. The newly added User-Defined Functions (UDF) and User-Defined Aggregate Functions (UDAFs) open the door for many novel use cases, for which you need to perform custom computations over your data when KSQL’s built-in scalar functions or built-in aggregation functions are insufficient. Applying a machine learning model in real time to a stream of data is an example of an exciting new use case this allows. UDFs take one input row to produce one output row (e.g., ABS, SUBSTRING), whereas UDAFs take numerous input rows to produce one output row (e.g., SUM, MAX, COUNT).

  KSQL also now supports stream-stream, table-table, and stream-table joins. Further, you can execute inner and outer joins in addition to left joins where appropriate for the joined entities. This means that KSQL now covers all of the available join operations in Kafka Streams, and this expanded scope lets you cover more use cases with KSQL. For example, in retail you can detect late shipments by joining a stream of orders with a stream of shipments; in advertising, you can match ad impressions with ad clicks.

  
  Using KSQL’s new INSERT INTO syntax, events from different source streams and queries can be written to the same output stream. This can be useful in various cases, including situations where you have the same logical data arriving on multiple physical topics.

  While users enjoy the ability to define stream processing logic simply through SQL-like queries, developers often want to interact with KSQL programmatically in order to compose and run such queries. Through the KSQL REST API, you can leverage your favorite programming language, whether it’s Python, Go, .NET, Java, JavaScript, and even shell scripting, to work with KSQL. This lets you compose KSQL queries at runtime and submit them to the server when you cannot know the query ahead of time, making KSQL and Confluent Platform an even better fit for non-Java virtual machine and/or polyglot environments.

  Streamline IoT integration with Kafka

  More and more customers of all industries are integrating their IoT devices and gateways with Apache Kafka using the MQTT standard for applications, as in the cases of connected cars, assembly lines in manufacturing, and predictive maintenance. Until today, project teams had to insert an MQTT broker into the middle as an intermediary before building a custom ingestion pipeline into Kafka.

  Confluent Platform 5.0 debuts MQTT Proxy to reduce these complexities and streamline IoT integration. It permits users to build a scalable proxy architecture similar to Confluent REST Proxy for HTTP(S) communication with Kafka. MQTT Proxy enables the replacement of third party MQTT brokers with Kafka-native MQTT proxies in order to remove the additional cost and complexity of intermediate storage and lag. The introduction of MQTT Proxy means that fewer services are required and the need to write a custom communication mechanism between MQTT and Kafka is eliminated.

  Using Transport Layer Security encryption and basic authentication, MQTT Proxy supports the widely used MQTT 3.1.1 protocol and allows publishing messages in all three MQTT quality of service levels. Future releases will add bidirectional communication and more security options.

  New in Apache Kafka 2.0

  Like previous releases, Confluent Platform 5.0 is built on the most recent version of Apache Kafka in order to provide customers with the latest in streaming technology. Confluent Platform 5.0 includes Apache Kafka 2.0, which contains a number of new features, performance improvements, bug fixes, and support changes, including:

  • 40 KIPs and 246 resolved JIRAs
  • Prefixed wildcard access control lists (ACLs), fine-grained ACLs for CreateTopics
  • Simple Authentication and Security Layer/OAuthBearer implementation
  • Improved quota communication and customization of quotas
  • Efficient memory usage for down conversion
  • Fix log divergence between leader and follower during fast leader failover
  • Improved timeout behaviour in the Java consumer, eliminating indefinite blocking when brokers are not available
  • Kafka Connect REST extension plugin, support for externalizing secrets, and improved error handling with dead-letter queue functionality
  • Scala API for Kafka Streams and other Streams API improvements

  Apache Kafka 2.0 also drops support for Java 7 and removes the previously deprecated Scala producer and consumer. More details are available in the Apache Kafka 2.0 Release Notes.

  Get started

  Here are the best ways to get started with Confluent Platform:

  • Try our free 30-day trial of Confluent Enterprise, which includes all of these additional capabilities.
  • Alternatively, you can install Confluent Open Source, which includes only open source components. Details for both Confluent Enterprise and Confluent Open Source can be found in the documentation.
  • Receive expert training and technical consulting to get up and running quickly. Confluent Enterprise 5.0 is backed by our subscription support.

  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!

  Special thanks to Apurva, Dan, Derrick, Erica, Ewen, Hojjat, Ismael, Jo, Joseph, Kai, Mikkin, Neha, Nick, Ran, Randall, Robin, Rohan, Rohit, Tim, Vahid, Victoria, and Vivian for their contributions to this post.

  The post Introducing Confluent Platform 5.0 appeared first on Confluent.

  
  
  https://www.confluent.io/blog/introducing-confluent-platform-5-0/
• 

  Highlights of the Kafka Summit San Francisco 2018 Agenda
  apache kafka kafka summit • • sorangutan  

  1
  1
  Posts
  106
  Views

  

  Honestly, it seems like I just filed my expense report for Kafka Summit London, and now we are announcing the program for Kafka Summit San Francisco. (CONFIDENTIAL TO CONFLUENT FINANCE: I did just file my expense report for Kafka Summit London. Three months isn’t so bad, right? Right?)

  But hey, here we are, just under three months out from the third annual marquee Apache Kafka^® community event. Kidding aside, Kafka Summit has established itself as the premier streaming systems event for people like you: developers, DevOps people, software architects, and anyone else who builds things with Kafka. Happily, people talk about Kafka in lots of different venues, but this is the place for the community to rally around our next-generation plans, share experiences, and plot the future of streaming technologies.

  Because Kafka has become an increasingly strategic platform, it has also attracted the attention of business leaders. To help welcome this new constituency, we are rolling out a new track at the 2018 summit: The Business Case.

  The Kafka Summit Program Committee just finished sorting through more than 250 excellent submissions and sent out formal acceptance emails last week. We’ll be publishing the full agenda (50+ sessions) soon, so keep an eye on the website or sign up for updates, and we’ll let you know as soon as the deed is done.

  Some highlights I can share from the agenda so far:

  Internals Track

  • Kafka Security 101 & Real-World Tips (Stephane Maarek, DataCumulus)
  • Connecting Kafka Across Multiple AWS VPCs (Benoit Carrière, Expedia)
  • Kafka on ZFS: Better Living Through Filesystems (Hugh O’Brien, Jet.com)

  Pipelines Track

  • Data Streaming Ecosystem Management at Booking.com (Alex Mironov, Booking.com)
  • More Data, More Problems: Scaling Kafka Mirroring Pipelines at LinkedIn (Celia Kung, LinkedIn)
  • Move the Needle of the Pin—Pinterest’s Story of Streaming Hundreds of Terabytes of Pins from MySQL to S3/Hadoop Continuously (Henry Cai, Pinterest)

  Streams Track

  • Deploying Kafka Streams Applications with Docker and Kubernetes (Gwen Shapira + Matthias J. Sax, Confluent)
  • How Big Fish Games Developed Real-Time Analytics Using Kafka Streams and Elasticsearch (Kalah Brown, Big Fish Games)
  • Real-Time Market Data Analytics Using Kafka Streams (Lei Chen, Bloomberg L.P.)

  The Business Case Track

  • Life is a Stream of Events (Bjørne Kvernstuen, Norwegian Directorate for Work and Welfare)
  • Kafka for Real-Time Event Processing in Serverless Environments (Jeff Sharpe, Capital One)
  • Using Kafka in Your Organization with Real-Time User Insights for a Customer 360 View (Chris Maier, Western Monroe Partners)

  The full speaker list will be posted soon.

  As always, permit me to thank the program committee for all their hard work and investment in putting together the agenda:

  • Becket Qin, LinkedIn
  • Ben Lorica, O’Reilly Media
  • Helena Edelson
  • Jun Rao, Confluent
  • Michael Noll, Confluent
  • Neha Narkhede, Confluent
  • Rajini Sivaram, Confluent
  • Sam Newman
  • Tim Berglund, Confluent
  • Todd Palino, LinkedIn

  Oh, hey one more thing: early bird pricing ends Aug. 1! “Oh, that’s August; I have plenty of time,” you think, but you and I both know that’s essentially tomorrow.

  Register now to save $200 off the standard conference pass.

  I look forward to seeing you at the summit!

  The post Highlights of the Kafka Summit San Francisco 2018 Agenda appeared first on Confluent.

  
  
  https://www.confluent.io/blog/highlights-of-the-kafka-summit-san-francisco-2018-agenda/
• 

  Ansible Playbooks for Confluent Platform
  confluent platform ansible playbooks • • sorangutan  

  1
  1
  Posts
  147
  Views

  

  Here at Confluent, we want to see each Confluent Platform installation be successful right from the start. That means going beyond beautiful APIs and spectacular stream processing features. We want to give engineers running the services a great experience, too, which is why I’m pleased to introduce our first set of Ansible playbooks to help set up and install Confluent Platform. These playbooks help you get a proof-of-concept cluster off the ground that is production-ready. To get started, just put your hostnames into the example hosts.yml file and run the full playbook with a simple `ansible-playbook -i hosts.yml all.yml`. The result is a secured Confluent Platform in less than five minutes.

  What makes these playbooks special? First, they deploy Confluent Platform services with the newly packaged systemd service unit files, giving administrators the familiar look and feel of a production system. We’ve also made the hard decisions for you by taking the liberties of setting well-documented default usernames and log locations to help installations of new environments be repeatable and recognizable. The systemd integration even provides a quite-lovely integration with journalctl for logging.

  The second feature is optional automated authentication and encryption setup. The supported modes are PLAINTEXT, SSL, and SASL_SSL. We often hear from users about how they want to get started quickly on a proof-of-concept to show the business value of a streaming data platform, but struggle to get past the first hurdle of enabling SASL and SSL. This setup generates simple SSL certificates and uses SASL authentication with a plaintext mechanism by  choosing the `sasl_ssl` template. The automated setup is designed to show the security capabilities of the platform simply and quickly. The best part is that when it’s time to go to full production, you can replace the provided self-signed SSL certificates and plaintext SASL with your own production grade and Kerberos principals without even having to worry if you missed a configuration.

  Finally, the setup isn’t limited to deploying some Confluent Platform services and leaving others as an exercise to the user. You can easily launch every component: Apache ZooKeeper, Kafka brokers, Confluent Schema Registry, Confluent REST Proxy, Kafka Connect workers, KSQL server, and Confluent Control Center. If you launch the whole platform, you also get the awesome topic and connector management features available in Confluent Control Center 4.1, and can get data flowing directly from the UI. In fact, you can check out Robin’s blog post for some great ideas on what to get connected to right away.

  The bottom line is, at Confluent, we want each of our users to be successful. Whether they are just getting started on a laptop, setting up their first proof-of-concept to show the business, or getting ready to launch their 15th production installation of the platform. We think that these playbooks are a great way to get those proof-of-concept installs off the ground in a way that follows best practices directly from our team.

  The repository is open! We would love to hear your feedback and welcome pull requests.

  Interested in more?

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

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

   

  The post Ansible Playbooks for Confluent Platform appeared first on Confluent.

  
  
  https://www.confluent.io/blog/ansible-playbooks-for-confluent-platform/
• 

  Apache Kafka® vs. Enterprise Service Bus (ESB)—Friends, Enemies, or Frenemies?
  apache kafka etl integration enterprise architecture enterprise service bus • • sorangutan  

  1
  1
  Posts
  96
  Views

  

  Typically, an Enterprise Service Bus (ESB) or other integration solutions like Extract-Transform-Load (ETL) tools have been used to try to decouple systems. However, the sheer number of connectors, as well as the requirement that applications publish and subscribe to the data at the same time, mean that systems are always intertwined. As a result, development projects have lots of dependencies on other systems and nothing can be truly decoupled.

  This blog post shows why so many enterprises leverage the open source ecosystem of Apache Kafka for successful integration of different legacy and modern applications, and how this differs but also complements existing integration solutions like ESB or ETL tools.

  The need for integration—a never ending story

  No matter in which enterprise you work, no matter when your company was founded, you will have the requirement to integrate your applications with each other to implement your business processes.

  This includes many different factors:

  • Technologies (standards like SOAP, REST, JMS, MQTT, data formats like JSON, XML, Apache Avro or Protocol Buffers, open frameworks like Nginx or Kubernetes, and proprietary interfaces like EDIFACT or SAP BAPI)
  • Programming languages and platforms like Cobol, Java, .NET, Go, or Python
  • Application architectures like Monolith, Client Server, Service-oriented Architecture (SOA), Microservices, or Serverless
  • Communication paradigms like batch processing, (near) real time, request-response, fire-and-forget, publish subscribe, continuous queries, and rewinding

  Many enterprise architectures are a bit messy—something like this:

  Every company needs to solve these spaghetti architectures. Depending on the decade, you either bought something like an ETL tool to build batch pipelines or an ESB to design a SOA. Some products also changed their names. Today, you are offered things like middleware messaging, an integration platform, microservice gateway, or API management. The branding and product name do not matter. You always see the same picture as a solution to move away from your spaghetti architecture to a central integral box in the middle, like this:

  

  This rarely worked well in practice, unfortunately. Most SOA projects in the last two decades failed. Instead of using an ETL tool or ESB for this, enterprises are now moving on to a streaming platform to solve this issue. Is this the next bubble on the market? Just a new term? Or, did something really change to allow successful integration across an enterprise—whether you integrate legacy mainframes, standard applications like CRM and ERPs, modern microservices built with any programming platform, or public cloud services? Why are companies now migrating to Apache Kafka to build this streaming platform? Why is everybody happy and talking about this at conferences, tech talks, and blog posts? How does it compare to an ESB or ETL tool?

  The next sections will answer all these questions, and explain the reason and differences between the open source ecosystem of Apache Kafka and other existing integration solutions.

  Event-driven processing and streaming as a key concept in the enterprise architecture

  A streaming platform (you can also enter another buzzword here) leverages events as a core principle. You think in data flows of events and process the data while it is in motion.

  Many concepts, such as event sourcing, or design patterns such as Enterprise Integration Patterns (EIPs), are based on event-driven architecture. The following are some characteristics of a streaming platform:

  • Event-based data flows as a foundation for (near) real-time and batch processing. In the past, everything was built on data stores (data at rest), making it impossible to build flexible, agile services to act on data while it is relevant.
  • Scalable central nervous system for events between any number of sources and sinks. Central does not mean one or two big boxes in the middle but a scalable, distributed infrastructure, built by design for zero downtime, handling the failure of nodes and networks, and rolling upgrades. Different versions of infrastructure (like Kafka) and applications (business services) can be deployed and managed in an agile, dynamic way.
  • Integrability of any kind of applications and systems. Technology does not matter. Connect anything: programming language, APIs like REST, open standards, proprietary tools, and legacy applications. Speed does not matter. Read once. Read several times. Read again from the beginning (e.g., add new application, train different machine learning models with the same data).
  • Distributed storage for decoupling applications. Don’t try to build your own streaming platform using your favorite traditional messaging system and in-memory cache/data grid. There is a lot of complexity behind this and a streaming platform simply has it built-in. This allows you to store the state of a microservice instead of requiring a separate database, for example.
  • Stateless service and stateful business processes. Business processes typically are stateful processes. They often need to be implemented with events and state changes, not with remote procedure calls and request-response style. Patterns like event sourcing and CQRS help implement this in an event-driven streaming architecture.

  Benefits of a streaming platform in the enterprise architecture

  A streaming platform establishes huge benefits for your enterprise architecture:

  • Large and elastic scalability regarding nodes, volume, throughput—all on commodity hardware, in any public cloud environments, or via hybrid deployments.
  • Flexibility of architecture. Build small services, big services, sometimes still even monoliths.  
  • Event-driven microservices. Asynchronously connected microservices model complex business flows and move data to where it is needed.
  • Openness without commitment to a unique technology or data format. The next new standard, protocol, programming language, or framework is coming for sure. The central streaming platform is open even if some sources or sinks use a proprietary data format or technology.
  • Independent and decoupled business services, managed as products, with their own lifecycle regarding development, testing, deployment, and monitoring. Loose coupling allows for independent speed of processing between different producers and consumers, on/offline modes, and handling backpressure.
  • Multi-tenancy to ensure that only the right user can create, write to, and read from different data streams in a single cluster.
  • Industrialized deployment using containers, devops, etc., deployed where needed, whether on premise, in the public cloud or in a hybrid environment.

  These characteristics build the foundation of a streaming platform, the beginning of your successful digital transformation. With services implementing a limited set of functions, and services being developed, deployed, and scaled independently, you get shorter time to results and increased flexibility. This is only possible with a streaming platform having the above characteristics.

  Use cases for a streaming platform

  Here are some generic scenarios for how you can leverage a streaming platform with the characteristics discussed above:

  • Event-driven processing of big data sets (e.g., logs, IoT sensors, social feeds)
  • Mission-critical, real-time applications (e.g., payments, fraud detection, customer experience)
  • Decoupled integration between different legacy applications and modern applications
  • Microservices architecture
  • Analytics (e.g., for data science, machine learning)

  Producer and consumers of different applications are really decoupled. They scale independently at their speed and requirements. You can add new applications over time, both on the producer and consumer side. Often, one event is required to be consumed by many independent applications to complete the business process. For example, a hotel room reservation needs immediate payment fraud detection in real time, the ability to process the booking through all backend systems in near real time, and overnight batch analytics to improve customer 360, aftersales, hotel logistics, and other business processes.

  While some processes need real-time processing, you also need to be capable of supporting batch processes. You even need re-consumption of data more often than you would think in the beginning, such as in cases of an application being down for some time, A/B testing with different versions of an application, adding a new application that needs to consume the data from scratch, or building different analytic models via machine learning based on the same data sets.

  Think about some more use cases that you can build easily with a real decoupled system that is still a well-integrated and scalable streaming platform:

  • Selling before the customer left the store
  • Aborting a transaction before the fraud happened
  • Replacing a part of a manufacturing machine before it breaks
  • Informing customers if a flight or train is late (plus sending updates, rebooking, or a voucher)   
  • You name it—the list goes on.

  Big bang from batch to real time?

  Now, you understand the added value of a real decoupled, scalable streaming platform. So, do I have to introduce this as a central data platform for all of our applications?

  Caution! No mature enterprise can do a big bang successfully. Legacy applications exist everywhere. Go step by step from pre-streaming to streaming platform. If you come from the mainframe ages, then you might even have batch and non-streaming applications forever (or realistically at least for the next 20-30 years). That’s fine. You just need to bring the events from these systems into the event-driven central nervous system.

  The following shows the streaming maturity model that we use to identify the current situation and planning in large enterprises:

  

  Where you are today?

  1. Pre-Streaming (Batch or legacy)
  2. Interest (First proof-of-concepts or pilots)
  3. Early Production (Some independent projects in production)
  4. Integrated Streaming (Streaming platform with different projects in production)
  5. Streaming Platform (Streaming enterprise with mostly event-based applications)

  Most traditional enterprises start their journey in the pre-streaming phase. That’s totally fine. The next section explains why almost any successful transformation into a streaming platform leverages the Apache Kafka ecosystem as a key architectural component.

  Introducing Apache Kafka ecosystem as streaming platform

  Often people are familiar with Apache Kafka, as it has been a hugely successful open source project, created at LinkedIn for big data log analytics. That was the beginning of Kafka, and just one of many use cases today. Kafka evolved from a data ingestion layer to a real-time streaming platform for all the use cases previously discussed. Many projects focus on building mission-critical applications around Kafka. It has to be up and performant 24/7. If Kafka is down, their business processes stop working.

  Kafka is unique because it combines messaging, storage, and processing of events all in one platform. It does this in a distributed architecture using a distributed commit log and topics divided into multiple partitions, as seen below:

  
  With this distributed architecture, Kafka is different from existing integration and messaging solutions. Not only is it massively scalable and built for high throughput but different consumers can also read data independently of each other and in different speeds.

  Applications publish data as a stream of events while other applications pick up that stream and consume it when they want. Because all events are stored, applications can hook into this stream and consume as required—in batch, real time or near-real-time. This means that you can truly decouple systems and enable proper agile development. Furthermore, a new system can subscribe to the stream and catch up with historic data up until the present before existing systems are properly decommissioned.

  The uniqueness of having messaging, storage, and processing in one distributed, scalable, fault-tolerant, high-volume, technology-independent streaming platform is the reason for the global success of Apache Kafka in almost every bigger company on this planet.

  So, what is Apache Kafka and its open source ecosystem? Let’s take a high-level look at its components:

  • Apache Kafka as core for distributed messaging and storage. High throughput, low latency, high availability, secure.
  • Kafka Connect as an integration framework for connecting external sources/destinations into Kafka.
  • Kafka Streams as a simple library that enables streaming application development within the Kafka framework.
  • Additional clients for non-Java programming languages, including C, C++, Python, .NET, Go, and several others.
  • Confluent REST Proxy to provide universal access to Kafka from any network connected device via HTTP(S).
  • Confluent Schema Registry as a central registry for the format of Kafka data, guaranteeing that all data is always consumable, including schema evolution.
  • Confluent KSQL as the streaming SQL engine that enables scalable, high-volume stream processing natively against Apache Kafka without writing source code.

  All these open source components are based on top of the core messaging and storage layer of Apache Kafka, and all leverage its features of high scalability, high volume/throughput, and failover. In addition, Confluent offers 24/7 support and enterprise tooling for end-to-end monitoring, management of Kafka clusters, multi-datacenter replication, and more.

  Okay, so we should start evaluating Kafka in a proof of concept (POC)?

  POC? Well, the Apache Kafka ecosystem is already battle-tested at a scale that you will probably not reach in the next 5-10 years. Many notable organizations are Kafka users:

  • LinkedIn processes over 4.5 trillion messages per day in 2018.
  • Netflix processes over 6 petabytes of data per day in 2018.
  • Ebay, Uber, Paypal, you-name-it. Just Google your favorite tech giant + Apache Kafka to see how they use Kafka. It is everywhere!
  • Even “traditional” companies in banking, insurance, telco, retail, automotive, and manufacturing use Kafka to transform and revolutionize their core business. Public references, including ING Group (banking), Audi (automotive), and Target (retail), to name a few, are available on the web.

  Don’t waste your time with POCs about performance, maturity, or scalability. Instead, evaluate how to integrate Kafka into your enterprise architecture to evolve business scenarios and innovate towards digital transformation.

  Should we replace our existing MQ and ESB deployments?

  Younger companies like Netflix, LinkedIn, and Zalando built their whole infrastructure on Kafka. Older companies are not that fortunate because they have plenty of mainframes, monoliths, and legacy technology. However, as discussed, a big bang replacement is not the right way to be successful. It’s a lot like transforming your home. Although it might make sense in theory to rebuild it from the ground up, oftentimes it is more practical to extend the house, change certain rooms, or redecorate.

  It’s difficult to just rip and replace mission-critical applications that are deeply integrated with other apps that work and involve a skilled team operating them. That said, sometimes it may be more cost effective to replace legacy architecture, just like it sometimes does make sense to remodel a house from the ground up. This was the case with Sberbank, the biggest bank in Russia, which built their complete core banking system around Kafka as central nervous system.

  Apache Kafka and other middleware as complementary components

  Legacy apps are typically based on complex data formats like EDIFACT, use complex interfaces like CORBA, or are built with an unmaintainable and inflexible programming language like Cobol. You cannot simply turn it off, or cut it out and replace it. This has to be done step by step. Legacy and modern applications co-exist to run the existing business and add new offerings to augment it.

  Innovate by integrating your old systems via a streaming platform—i.e., the Apache Kafka ecosystem. Use concepts like change data capture (CDC) and integration tools, such as an ESB or ETL, with great graphical tooling and connectors for legacy applications. With this foundation, you can build new applications with modern technologies, big data systems, machine learning, etc., natively around Apache Kafka, and at the same time keep access to your legacy events, which you still need for added value in new projects.

  Real decoupling and technology independence also means that you have dumb pipes and smart endpoints. If you build all the integration logic (or, even worse, some business logic) into the central integration layer, then all your scalability, agility, and independence of the different systems are gone. This is a key difference to traditional integration solutions in which you put all the logic in the middle layer ESB. It creates dependency on this (proprietary) technology/API, as well as inflexibility.

  Smart endpoints can be anything. You can leverage the Kafka ecosystem to build applications around Kafka with Kafka Streams, KSQL, or any Kafka client like Java, .Net, Python, or Go. You can also use any other application to integrate other applications with Kafka. The secret to long-term success is that the infrastructure is open to any technology and architectural pattern.

  ETL and ESB have excellent tooling, including graphical mappings for doing complex integration with SOAP, EDIFACT, SAP BAPI, COBOL, etc. (Trust me, you don’t want to write the code for this.) It is already running, paid, and integrated. Therefore, existing MQ and ESB solutions, which already integrate with your legacy world, are not competitive to Apache Kafka. Rather, they are complementary! Leverage them like you did in the past to integrate with the old world. You can use the following:

  • Confluent JMS connector for IBM MQ and other JMS brokers
  • Kafka Connect CDC connectors for mainframe and databases
  • ESB or ETL tools, which integrate with legacy protocols (SOAP, EDIFACT, etc.) and applications (SAP, Siebel, etc.). All these tools also have Kafka connectors in the meantime.
  • Traditional messaging brokers also have a proxy to Kafka in the meantime. Even though some time ago they told everybody that you cannot build a universal messaging platform, traditional messaging brokers now integrate to Kafka and other standards like MQTT to not become forgotten
  • Kafka’s official low-level client APIs like Java, .NET, Go, and Python implement a direct integration if no other more feasible integration option is available.

  As a last note, after explaining the usage of ESB and ETL tools in the Kafka world, I quote ThoughtWorks to remind you not to try building a new ESB around Kafka:

  Some organizations recreate ESB antipatterns with Kafka by centralizing the Kafka ecosystem components—such as connectors and stream processors—instead of allowing these components to live with product or service teams. This reminds us of seriously problematic ESB antipatterns, where more and more logic, orchestration and transformation were thrust into a centrally managed ESB, creating a significant dependency on a centralized team. We’re calling this out to dissuade further implementations of this flawed pattern.

  Source: https://www.thoughtworks.com/radar/techniques/recreating-esb-antipatterns-with-kafka

  Apache Kafka and its ecosystem is designed as a distributed architecture with many smart features built-in to allow high throughput, high scalability, fault tolerance and failover! Let the product or service teams build their applications with Kafka Streams, KSQL, and any other Kafka client API. Integrate Kafka with ESB and ETL tools if you need their features for specific legacy integration. An ESB or ETL process can be a source or sink to Apache Kafka like any other Kafka producer or consumer API. Oftentimes, the integration with legacy systems using such a tool is built and running already anyway. Currently all these tools also have a Kafka connector because the market drives them this way. So, you just need to combine the existing integration with the Kafka connector, and there you have it: flexible, scalable and highly available integration between legacy and future ecosystems through Kafka.

  Apache Kafka as streaming platform between legacy and the new modern world

  Apache Kafka is an open source streaming platform that allows you to build a scalable, distributed infrastructure that integrates legacy and modern applications in a flexible, decoupled way. It is already battle-tested for processing trillions of messages and petabytes of data per day. Simply leverage the Apache Kafka ecosystem in your enterprise architecture to make integration of your various systems successful and dynamic. But whatever you do, do not try to build an ESB around Kafka—it is an anti-pattern that will create inflexibility and unwanted dependencies. Instead, leverage the distributed architecture of the Apache Kafka ecosystem to build a flexible, event-driven streaming infrastructure with high throughput, high scalability, fault tolerance, and failover.

  As a next step, you can get further proof by looking at past Kafka Summits in London, New York, and San Francisco, where plenty of tech giants, established companies, and software vendors explained how they digitize their business and innovate their products leveraging the Apache Kafka ecosystem. All video recordings and slide decks are available online.

  Interested in more?

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

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

  The post Apache Kafka® vs. Enterprise Service Bus (ESB)—Friends, Enemies, or Frenemies? appeared first on Confluent.

  
  
  https://www.confluent.io/blog/apache-kafka-vs-enterprise-service-bus-esb-friends-enemies-or-frenemies/
• 

  June Preview Release: Packing Confluent Platform with the Features You Requested!
  ksql confluent platform confluent control center preview release • • sorangutan  

  1
  1
  Posts
  76
  Views

  

  We are very excited to announce Confluent Platform June 2018 Preview. This is our most feature-packed preview release for Confluent Platform since we started doing our monthly preview releases in April 2018.

  Read on to learn more, and remember to share your feedback to help improve the Apache Kafka ecosystem. You can do that by visiting the Confluent Community Slack channel (particularly the #ksql and #control-center channels) or by contributing to the KSQL project on GitHub, where you can file issues, submit pull requests, and contribute to discussions.

  Download The Preview Release

   

  Confluent Control Center

  Confluent Schema Registry Support for Topic Inspection

  Confluent Control Center’s topic inspection feature already supports JSON and string message formats, but starting with this release, it will have full support for Avro data through the Confluent Schema Registry.

  Schema Registry, as we all should know, is a component of Confluent Platform that stores a versioned history of all schemas and works with clients to support predictable schema evolution. To configure Schema Registry in Control Center, simply add confluent.controlcenter.schema.registry.url=<your_schema_registry_server> in control-center.properties file before starting up the application. The message viewer in the topic inspection tab will automatically detect and deserialize Avro records.

  

  Feature Access Control

  In this release, we’re also introducing feature access toggles within Confluent Control Center for KSQL, Topic Inspection, and Schema Registry. In many deployments, the end users of Control Center are not allowed to view records in topics for security and compliance reasons. In order to allow admins to control application-wide access to features that reveal topic data in Control Center, we added the option to control access at start-up time. When you restrict access to a feature, Control Center’s UI will reflect this change, and users cannot circumvent these protections in any way.

  To enable/disable a feature, change the following configs in the Control Center properties file before starting up:

  confluent.controlcenter.ksql.enable=[true|false]

  confluent.controlcenter.schema.registry.enable=[true|false]

  confluent.controlcenter.topic.inspection.enable=[true|false]

   

  KSQL

  Working with Nested Data

  One of the most requested enhancements for KSQL has been support for complex nested data types. Starting with the June preview release, you can now declare streams and tables with columns using a STRUCT data type in your CREATE STREAM and CREATE TABLE statements, and then access the internal fields of these columns in your SELECT queries as you do in any other expression. STRUCT support is available for both JSON and Avro data formats. Nested fields in the STRUCT can be any valid type in KSQL, including STRUCT,  ARRAY and MAP. For instance, you can have a column that is an ARRAY of STRUCTs, where one of the fields in the STRUCT is a MAP. This means KSQL now supports querying data with many nested levels.

  The following is an example of declaring an ORDERS stream, where the ADDRESS column has a STRUCT type with nested fields for CITY, STATE, and ZIPCODE:

  
  You can refer to fields in a STRUCT using the `->` operator:

   

  Beyond SQL with User Defined Functions

  Another highly requested feature for KSQL has been the ability to define your own functions that go beyond what’s provided by KSQL out of the box. User Defined Functions (UDFs) and User Defined Aggregate Functions (UDAFs) open the door for many novel use cases where you need to perform custom computations over your data. UDFs take one input row to produce one output row (e.g., ABS and SUBSTRING) whereas UDAFs take n input rows to produce one output row (e.g., SUM, MAX, COUNT).

  Using the new UDF and UDAF features,you can define custom computations and make them available to your KSQL queries in case the built-in scalar functions or the built-in aggregation functions are not sufficient.

  For now, UDF(A)Fs need to be implemented in Java. You simply write the function code and mark the classes with @Udf or @UdafFactory annotations. To make the UDF(A)Fs available to KSQL, you must create an uberjar of the UDF(A)Fs code including any transitive dependencies, and then place the jar file in the `ext/` folder of a KSQL Server. To better secure your KSQL deployment environment when using UDF(A)Fs you can also provide an optional blacklisting configuration that will prevent the use of certain Java classes and packages such as `java.io.*`. Finally, you can use the SHOW FUNCTIONS statement to get a listing of all available functions including UDFs and UDAFs.

  Here is an example UDF that returns the length of a string:

  
  Here is an example of calling the `STRING_LENGTH` function:

  
  Even more interesting is the ability to bring your own custom aggregation functions. Here we define a UDAF for computing the sum of string lengths:

  
  This UDAF can then be used in your KSQL queries:

   

  Stream-Stream and Table-Table Joins

  Beyond the existing Stream-Table joins, the June preview release introduces Stream-Stream and Table-Table joins.  For each of these joins, KSQL supports inner, full outer, and left join types. This means KSQL now covers all of the available join operations in Kafka Streams.

  A Stream-Stream join enables KSQL users to join two streams based on the desired message key within a given join window. Consider the following scenario where we have two streams, ORDERS and SHIPMENTS. Assuming that an order is late if it is shipped more than 2 hours after the order was placed, we can create a new stream for the late orders by joining the two streams and filtering the orders that were not shipped in the given time window. The following query creates the stream of late orders:

  The above query will match every shipment row with the order rows that are within the 2-hour window. If there is no match, the right-hand side of the join result will be NULL, indicating that the given order was not shipped within the expected time. Note that we used the `WITHIN` keyword in the Stream-Stream join to specify the join window.  

  Where to go from here

  Try out the new Confluent Platform June 2018 Preview release and share your feedback! Here’s what you can do to get started:

  • Download Confluent Platform June 2018 preview release.
  • Check out the Docker images for KSQL Server and KSQL CLI.
  • Visit the Confluent Community Slack channel to ask questions or share feedback. To discuss KSQL, visit the #ksql channel. For Control Center, visit the #control-center channel.
  • If you are interested in contributing to KSQL, visit the KSQL project on GitHub. Feel free to file issues, submit pull requests, and contribute to discussions there. Remember that KSQL is open source and already includes some great community contributions – the next one could be yours!

   

  The post June Preview Release: Packing Confluent Platform with the Features You Requested! appeared first on Confluent.

  
  
  https://www.confluent.io/blog/june-preview-release-confluent-plaform/
• 

  We ❤️ syslogs: Real-time syslog processing with Apache Kafka and KSQL—Part 3: Enriching events with external data
  ksql apache kafka stream processing mongodb kafka connect syslog • • sorangutan  

  1
  1
  Posts
  139
  Views

  

  Using KSQL, the SQL streaming engine for Apache Kafka®, it’s straightforward to build streaming data pipelines that filter, aggregate, and enrich inbound data. The data could be from numerous sources, including applications, log files, databases, and IoT and network devices. In the two previous articles (1 | 2) in this series I’ve shown how to use Apache Kafka and its Connect API to ingest syslog data from multiple sources. Using KSQL, it’s easy to filter it as well as use aggregations to drive simple anomaly detection. The output from KSQL stream processing is a Kafka topic, which can you stream with Kafka Connect to a data store such as S3, or use to drive applications directly such as push-notification alerting.

  

  In this article we’re going to conclude our fun with syslog data by looking at how we can enrich inbound streams of syslog data with reference information from elsewhere to produce a real-time enriched data stream. The syslog data in this example comes from various servers and network devices, and the additional information with which we’re going to enrich it is from MongoDB, which happens to be the datastore used by Ubiquiti network devices. With the enriched data we’re going to drive some real-time analytics through Elasticsearch and Kibana, as well as trigger push notifications based on activity of certain devices on the network.

  Streaming syslog data from Ubiquiti devices

  

  I’m a big fan of Ubiquiti’s network devices, including their Unifi access points. I could wax lyrical here as to why, but I will instead refer you to Troy Hunt’s thorough write-up of Ubiquiti. One of the nice characteristics of them is that there is a ton of configuration that you can do to do them, of the sort that typical consumer-grade networking equipment just doesn’t support.

  Configuring the Unifi controller to write syslog data for all devices is easily done through the GUI:

  

  The syslog server and port should be those where you’re running the Kafka Connect syslog agent (see here for setup instructions). Run the console consumer against the topic to validate that you’re getting data:

  $ kafka-avro-console-consumer \
   --bootstrap-server localhost:9092 \
   --property schema.registry.url=http://localhost:8081 \
   --topic syslog
  
  {"date":{"long":1522059085000},"facility":{"int":3},"host":{"string":"(\"U7PG2,xxxxxxxxxx,v3.7.40.6115\")"},"level":{"int":6},"message":{"string":"(\"U7PG2,xxxxxxxxxx,v3.7.40.6115\") hostapd: ath3: STA xx:xx:xx:xx:xx:xx IEEE 802.11: associated\n"},"charset":{"string":"UTF-8"},"remote_address":{"string":"/192.168.10.71:45584"},"hostname":{"string":"192.168.10.71"}}

  Working with Ubiquiti syslog data in Apache Kafka and KSQL

  From the KSQL prompt, we can inspect the topic, using PRINT:

  ksql> PRINT 'syslog';
  Format:AVRO
  3/4/18 9:14:35 AM GMT, /(/192.168.10.87:50442, {"date": 1520154874000, "facility": 3, "host": "asgard02", "level": 5, "message": "asgard02 syslogd[4134]: --- syslogd restarted ---\n", "charset": "UTF-8", "remote_address": "/192.168.10.87:50442", "hostname": "192.168.10.87"}
  3/4/18 9:14:35 AM GMT, /(/192.168.10.87:50442, {"date": 1520154874000, "facility": 5, "host": "asgard02", "level": 5, "message": "asgard02 syslogd[4134]: Configuration Notice:\n\tASL Module \"com.apple.cdscheduler\" claims selected messages.\n\tThose messages may not appear in standard system log files or in the ASL database.\n", "charset": "UTF-8", "remote_address": "/192.168.10.87:50442", "hostname": "192.168.10.87"}

  You’ll note that KSQL determines automagically that the data is in Avro format, and displays the timestamp, key, and message value. Declaring a KSQL stream on top of these syslog events is simple:

  ksql> CREATE STREAM SYSLOG WITH (KAFKA_TOPIC='syslog',VALUE_FORMAT='AVRO');

  Looking at the data some more, we can inspect the host names:

  ksql> SELECT HOST FROM SYSLOG WHERE HOST !='asgard02' LIMIT 50;
  ("U7PG2,f09fc2000000,v3.7.40.6115")
  rpi-03
  proxmox01
  asgard02
  ("BZ2,24a43c000000,v3.7.40.6115")

  Whilst most devices on my network use a standard hostname when sending their syslog data, Ubiquiti ones send a hostname that looks like this:

  (\"U7PG2,f09fc2000000,v3.7.40.6115\")

  Where this appears to be a concatenation of device model, MAC, and firmware version. I can use this ‘fingerprint’ to pick out all syslog traffic relating to these devices in KSQL:

  ksql> SELECT MESSAGE FROM SYSLOG WHERE HOST LIKE '(%';
  ("U7PG2,f09fc2000000,v3.7.40.6115") hostapd: ath0: STA 60:01:94:xx:xx:xx IEEE 802.11: disassociated
  ("U7PG2,f09fc2000000,v3.7.40.6115") kernel: [7142078.740000]  ieee80211_ioctl_kickmac[14643]
  ("U7PG2,f09fc2000000,v3.7.40.6115") libubnt[9577]: dpi.dpi_stainfo_notify(): dpi not enable
  ("U7PG2,f09fc2000000,v3.7.40.6115") libubnt[9577]: wevent.ubnt_custom_event(): EVENT_STA_LEAVE ath0: 60:01:94:xx:xx:xx / 0

  (NB: a better pattern would be (%,%,%) but KSQL currently only supports single wildcard LIKE predicates)

  

  Let’s persist this stream of Ubiquiti syslog data to make it easier to work with in subsequent querying and processing:

  ksql> CREATE STREAM UBNT_SYSLOG AS SELECT * FROM SYSLOG WHERE HOST LIKE '(%';
  
   Message
  ----------------------------
   Stream created and running
  ----------------------------
  ksql> DESCRIBE UBNT_SYSLOG;
  
   Field          | Type
  --------------------------------------------
   ROWTIME        | BIGINT           (system)
   ROWKEY         | VARCHAR(STRING)  (system)
   DATE           | BIGINT
   FACILITY       | INTEGER
   HOST           | VARCHAR(STRING)
   LEVEL          | INTEGER
   MESSAGE        | VARCHAR(STRING)
   CHARSET        | VARCHAR(STRING)
   REMOTE_ADDRESS | VARCHAR(STRING)
   HOSTNAME       | VARCHAR(STRING)
  --------------------------------------------
  For runtime statistics and query details run: DESCRIBE EXTENDED <Stream,Table>;
  ksql>

  From this stream, we can examine our data and look for certain conditions and events—using standard SQL predicates. For example, we can see devices connecting to an Access Point:

  ksql> SELECT MESSAGE from UBNT_SYSLOG WHERE MESSAGE LIKE '% associated%';
  ("U7PG2,f09fc2000000,v3.7.40.6115") hostapd: ath3: STA 38:f7:3d:xx:xx:xx IEEE 802.11: associated

  So this in itself it pretty neat, as we can filter in realtime the syslog data that’s coming in. But…how about being able to identify the access point and user device names? At the moment we can only see their MAC addresses. Read on to see how!

  Accessing Ubiquiti data from MongoDB via KSQL and Debezium

  

  To get additional metadata from our network devices, including things like the access point and user device names, we’re going to stream this data from its source into Kafka. Ubiquiti uses MongoDB to hold this data, and there’s a great change-data-capture (CDC) tool built on Kafka Connect called Debezium that supports several source databases—including MongoDB! We can use this to stream the current contents of the database into a Kafka topic, as well as all and every subsequent changes to that data. This means that in effect we maintain a live mirror of our MongoDB database in Kafka itself. We don’t need to do some kind of convoluted lookups from Kafka out to a database each time we want to look up some reference data, since we have now that reference data within Kafka itself.

  For the purposes of this blog, I cloned my MongoDB instance from the Ubiquiti controller onto a separate server. For a real-life example then we’d use the live controller itself – since working against a clone means that any new changes to the data (such as new access points being created) won’t be available to us.

  You can use a tool such as Robo 3T to explore the data that Ubiqiuti has within it:

  

  Setting up Debezium to stream the data from MongoDB is straightforward – you can follow the steps in this guide. After doing it, you’ll have a bunch of Kafka topics, each reflecting the data in each of the MongoDB collections. The two that we’re interested in for this article are:

  • device – metadata for access points
  • user – metadata for user devices (i.e. stuff that connects to the access points—phones, computers, etc)

  Let’s take a look at the data we’ve brought in, using KSQL. Create a stream over the inbound Device data topic (which is metadata about the access points), and use the EXTRACTJSONFIELD function to show specific fields:

  ksql> CREATE STREAM DEVICE_RAW (after VARCHAR) WITH (KAFKA_TOPIC='ubnt.ace.device',VALUE_FORMAT='AVRO');
  ksql> SELECT EXTRACTJSONFIELD(after,'$.name'),EXTRACTJSONFIELD(after,'$.ip') FROM DEVICE_RAW;
  Unifi AP - Study | 192.168.10.68
  Unifi AP - Attic | 192.168.10.67
  ubnt.moffatt.me | 77.102.5.159
  Unifi AP - Pantry | 192.168.10.71

  Now let’s declare all the columns of interest in our schema, and using CREATE STREAM AS SELECT (“CSAS”) generate an Avro topic based on the source stream. This resulting topic will hold not only the transformed data that’s currently on the source topic from MongoDB, but also any subsequent changes to that data.

  ksql> CREATE STREAM UBNT_DEVICE_SRC AS \
  SELECT EXTRACTJSONFIELD(after,'$._id') as id, \
  EXTRACTJSONFIELD(after,'$.name') as name, \
  EXTRACTJSONFIELD(after,'$.ip') as ip, \
  EXTRACTJSONFIELD(after,'$.mac') as mac, \
  EXTRACTJSONFIELD(after,'$.model') as model, \
  EXTRACTJSONFIELD(after,'$.version') as version \
  FROM DEVICE_RAW;
  
  ksql> SELECT id, name, ip, mac, model, version FROM UBNT_DEVICE_SRC;
  {"$oid":"58385328e4b001431e4e497a"} | Unifi AP - Study | 192.168.10.68 | xx:xx:xx:xx:xx:xx | BZ2 | 3.7.40.6115
  {"$oid":"583854cde4b001431e4e4982"} | Unifi AP - Attic | 192.168.10.67 | xx:xx:xx:xx:xx:xx | BZ2 | 3.7.40.6115
  {"$oid":"58b3fb48e4b0b79e50242621"} | ubnt.moffatt.me | 77.102.5.159 | xx:xx:xx:xx:xx:xx | UGW3 | 4.3.34.4943823
  {"$oid":"58b406f1e4b0e334d74c46e4"} | Unifi AP - Pantry | 192.168.10.71 | xx:xx:xx:xx:xx:xx | U7PG2 | 3.7.40.6115

  Because we want to join this data (which is, in effect, reference data) we need to ensure that the topic’s messages are keyed on the column on which we want to join, which is the IP address.

  Let’s see what the key currently is, courtesy of the system column ROWKEY:

  ksql> SELECT ROWKEY FROM UBNT_DEVICE_SRC;
  {"id":"{ \"$oid\" : \"58b3fb48e4b0b79e50242621\"}"}
  {"id":"{ \"$oid\" : \"58385328e4b001431e4e497a\"}"}
  {"id":"{ \"$oid\" : \"58b406f1e4b0e334d74c46e4\"}"}

  The source topic is keyed on the id field, which traces back to the source MongoDB collection. We want to rekey the topic to use ip instead. KSQL is a simple way to do this, using the PARTITION BY clause:

  ksql> CREATE STREAM UBNT_DEVICE_SRC_REKEY AS SELECT * FROM UBNT_DEVICE_SRC PARTITION BY IP;
  
   Message
  ----------------------------
   Stream created and running
  ----------------------------

  Let’s verify that the message key is now the same as the ip field in the message value:

  ksql> SELECT ROWKEY, ip FROM UBNT_DEVICE_SRC_REKEY;
  192.168.10.68 | 192.168.10.68
  77.102.5.159 | 77.102.5.159
  192.168.10.67 | 192.168.10.67

  Perfect! The final part of this is to take the resulting Kafka topic from this KSQL streaming transformation and use it as the basis for our KSQL table which will be joined to the syslog data.

  ksql> CREATE TABLE UBNT_DEVICE WITH (KAFKA_TOPIC='UBNT_DEVICE_SRC_REKEY',VALUE_FORMAT='AVRO',KEY='IP');

  It’s crucial that our table’s topic messages are keyed on the join column which we will be using, so let’s verify again that it is indeed the case in our new table:

  ksql> SELECT ROWKEY, IP, NAME FROM UBNT_DEVICE;
  192.168.10.68 | 192.168.10.68 | Unifi AP - Study
  192.168.10.71 | 192.168.10.71 | Unifi AP - Pantry
  192.168.10.67 | 192.168.10.67 | Unifi AP - Attic
  77.102.5.159 | 77.102.5.159 | ubnt.moffatt.me

  Q: Why did we create a STREAM of device data, and then a TABLE? When should we use STREAM and when should we use TABLE? A: Great question! Logically, we are using the device data as a TABLE. That is, we want to join an inbound stream of events to our device data in order to enrich it. We want to know for a given key, what the corresponding values are. For a given device row, what’s its name, it’s model, it’s version etc. So it is most definitely a TABLE. But, in order to join to a TABLE, that TABLE must be keyed on the join column. And as we saw from inspecting ROWKEY above, this was not the case. So we utilised KSQL’s powerful re-keying functionality to rekey the topic automagically. And to do that, we treat the inbound data as a STREAM. Why? Because it’s simply an inbound Kafka topic of events, partitioned on one column and on which we want to partition another. Each event (in this context, a change to the source devicedata on MongoDB) simply needs re-routing to the output topic with the new partitioning key.

  Joining Ubiquiti syslog events with access point reference data

  

  Having defined our inbound stream of Ubiquiti events (captured through Kafka Connect’s syslog connector), and device reference data sourced from MongoDB (snapshotted and changes both streamed through Kafka Connect from Debezium), we can join the two, which is remarkably straightforward to anyone familiar with SQL. Here we’re showing each syslog message, enriched with the name of the access point from which it originated:

  ksql> SELECT D.NAME, L.MESSAGE \
  FROM UBNT_SYSLOG L \
  LEFT JOIN UBNT_DEVICE D \
  ON L.HOSTNAME=D.IP;
  
  Unifi AP - Pantry | ("U7PG2,f09fc2000000,v3.7.40.6115") kernel: [5255656.050000] ieee80211_ioctl_set_ratelimit: node with aid 8 and mac cc:2d:b7:xx:xx:xx has been tagged non rate-limiting
  Unifi AP - Attic | ("BZ2,dc9fdbxxxxxx,v3.7.40.6115") hostapd: ath1: STA 3c:2e:f9:xx:xx:xx IEEE 802.11: associated
  Unifi AP - Attic | ("BZ2,dc9fdbxxxxxx,v3.7.40.6115") syslog: dpi.dpi_stainfo_notify(): dpi not enable

  In the syslog data from Ubiquiti is a whole wealth of events, many of them more low-level than we may be interested in. One of the things that is useful to track is user devices connecting to access points, and we can easily expose this here using a SQL predicate. We can also expose the timestamp at which this happened, made human-readable with the TIMESTAMPTOSTRING function:

  ksql> SELECT TIMESTAMPTOSTRING(L.ROWTIME, 'yyyy-MM-dd HH:mm:ss') AS CONNECT_TS, D.NAME, L.MESSAGE \
  FROM UBNT_SYSLOG L \
  LEFT JOIN UBNT_DEVICE D \
  ON L.HOSTNAME=D.IP \
  WHERE L.MESSAGE LIKE '% associated%';
  
  2018-03-28 09:37:33 | Unifi AP - Attic | ("BZ2,dc9fdbxxxxxx,v3.7.40.6115") hostapd: ath1: STA 3c:2e:f9:xx:xx:xx IEEE 802.11: associated
  2018-03-27 17:15:11 | Unifi AP - Study | ("BZ2,24a43cde91a0,v3.7.40.6115") hostapd: ath1: STA fc:a1:83:xx:xx:xx IEEE 802.11: associated

  So this is pretty cool, as we can now see as events occur more information about them—in this case the access point that a device is joining. Let’s persist it as a stream, and add in the derivation of the MAC address of the connecting device, extracted using the SUBSTRING function:

  ksql> CREATE STREAM UBNT_SYSLOG_AP_CONNECTS AS \
  SELECT D.NAME AS AP_NAME, \
  SUBSTRING(L.MESSAGE,53,70) as USER_MAC \
  FROM UBNT_SYSLOG L \
  LEFT JOIN UBNT_DEVICE D \
  ON L.HOSTNAME=D.IP \
  WHERE L.MESSAGE LIKE '% associated%';
  
   Message
  ----------------------------
   Stream created and running
  ----------------------------
  ksql> SELECT TIMESTAMPTOSTRING(ROWTIME, 'yyyy-MM-dd HH:mm:ss') , AP_NAME, USER_MAC \
  FROM UBNT_SYSLOG_AP_CONNECTS;
  2018-03-28 09:37:33 | Unifi AP - Attic | 3c:2e:f9:xx:xx:xx
  2018-03-27 17:15:11 | Unifi AP - Study | fc:a1:83:xx:xx:xx

  But what about the user device—the third column in the above output? All we have is a MAC address (obfuscated in the examples above). Wouldn’t it be nice to see the name of the device as well as the access point? Well, you know what’s coming now!

  Joining Ubiquiti syslog events with access point and device reference data

  

  In the same way as we saw above how to manipulate the inbound stream of snapshotted device (access point) data and subsequent changes (streamed from MongoDB via Debezium), we’ll do the same here for the device information, which is found in the user collection. The steps are:

  1. Declare a raw STREAM on source user MongoDB collection data, in which all data resides in the after field

     ksql> CREATE STREAM USER_RAW (AFTER VARCHAR) WITH (KAFKA_TOPIC='ubnt.ace.user',VALUE_FORMAT='AVRO');
     
      Message
     ----------------
      Stream created
     ----------------

  2. Declare a derived STREAM in which columns are extracted from the source after field

     ksql> CREATE STREAM UBNT_USER_SRC AS SELECT EXTRACTJSONFIELD(after,'$.mac') as mac, \
     EXTRACTJSONFIELD(after,'$.oui') as oui, \
     EXTRACTJSONFIELD(after,'$.name') as name, \
     EXTRACTJSONFIELD(after,'$.hostname') as hostname, \
     EXTRACTJSONFIELD(after,'$.is_guest') as is_guest \
     FROM USER_RAW;
     
      Message
     ----------------------------
      Stream created and running
     ----------------------------

  3. Re-key the STREAM by join column (MAC address)

     ksql> CREATE STREAM UBNT_USER_SRC_REKEY AS SELECT * FROM UBNT_USER_SRC PARTITION BY MAC;
     
      Message
     ----------------------------
      Stream created and running
     ----------------------------

  4. Declare a TABLE on top of the Kafka topic resulting from the re-key operation

     ksql> CREATE TABLE UBNT_USER WITH (KAFKA_TOPIC='UBNT_USER_SRC_REKEY',VALUE_FORMAT='AVRO',KEY='MAC');
     
      Message
     ---------------
      Table created
     ---------------

  5. Check that the message key (ROWKEY) matches the intended join column (MAC):

     ksql> SELECT ROWKEY, MAC, NAME FROM UBNT_USER;
     b8:27:eb:xx:xx:xx | b8:27:eb:xx:xx:xx | rpi-01.moffatt.me
     ac:bc:32:xx:xx:xx | ac:bc:32:xx:xx:xx | asgard
     1e:7e:60:xx:xx:xx | 1e:7e:60:xx:xx:xx | media02.moffatt.me
     fc:a1:83:xx:xx:xx | fc:a1:83:xx:xx:xx | Fire 01 (Red)
     60:01:94:xx:xx:xx | 60:01:94:xx:xx:xx | Attic lights plug
     00:04:20:xx:xx:xx | 00:04:20:xx:xx:xx | Squeezebox - Kitchen
     b8:ac:6f:xx:xx:xx | b8:ac:6f:xx:xx:xx | crashplan.moffatt.me

  Now we can join between Ubiquiti syslog events and reference information for both access points (persisted above in the UBNT_SYSLOG_AP_CONNECTS stream), and user devices (UBNT_USER):

  ksql> SELECT TIMESTAMPTOSTRING(L.ROWTIME, 'yyyy-MM-dd HH:mm:ss') AS CONNECT_TS, \
  L.AP_NAME, U.NAME AS USER_DEVICE_NAME , U.HOSTNAME as HOSTNAME, U.IS_GUEST  \
  FROM UBNT_SYSLOG_AP_CONNECTS L \
  LEFT JOIN UBNT_USER U \
  ON L.USER_MAC = U.MAC ;
  2018-06-11 15:05:06 | Unifi AP - Study | null | ItStillnsWhenIP | false
  2018-06-11 15:05:06 | Unifi AP - Study | null | ItStillnsWhenIP | false
  2018-06-11 15:13:46 | Unifi AP - Study | Fire 02 (Yellow) | amazon-ca6091cfe | false

  So every time a user’s device connects to an access point, we get to see the name of the access point, the name of the user device, and the type of the device. We’ll persist this as a stream, because we’ve not quite finished this exploration yet!

  ksql> CREATE STREAM UBNT_AP_USER_DEVICE_CONNECTS AS \
  SELECT TIMESTAMPTOSTRING(L.ROWTIME, 'yyyy-MM-dd HH:mm:ss') AS CONNECT_TS, \
  L.AP_NAME, U.NAME AS USER_DEVICE_NAME , U.HOSTNAME as HOSTNAME, U.IS_GUEST  \
  FROM UBNT_SYSLOG_AP_CONNECTS L \
  LEFT JOIN UBNT_USER U \
  ON L.USER_MAC = U.MAC ;

  

  Taking this enriched stream of data we can use a tool such as Elasticsearch with Kibana on top to provide an easy visualisation of the real-time data, as well as aggregate analysis based upon it:

  

  Streaming the Kafka topic to Elasticsearch is easy using the Kafka Connect Elasticsearch plugin. Take a look at the Elasticsearch mapping template and Kafka Connect configuration I am using.

  Enriched streams of data are valuable for analysis that we want to consume and look at, but even more valuable is event-driven alerting on conditions that we’re interested in. We saw in the previous article how we can use something like a simple Python script to drive push-based notifications in response to events on a Kafka topic. Using this pattern we can use a further KSQL expression to send notifications to a topic when a connection is made to an access point from a certain type of device. For example, if I wanted to track whenever my wifi-enabled plugs (yes, really!) reconnect to an access point, I can filter on the device type of “Espressi”:

  ksql> SELECT CONNECT_TS, AP_NAME, USER_DEVICE_NAME \
  FROM UBNT_AP_USER_DEVICE_CONNECTS \
  WHERE DEVICE_TYPE='Espressi';
  2018-03-04 14:07:43 | Unifi AP - Study | Wifi Plug - Sitting Room
  2018-03-04 14:07:48 | Unifi AP - Study | Attic lights plug
  2018-03-04 14:07:42 | Unifi AP - Study | Wifi Plug - Sitting Room
  2018-03-04 14:07:42 | Unifi AP - Study | Study light

  

  Persisting this to a target stream that the python application is listening to results in a nice push notification every time the device reconnects…

  CREATE STREAM KSQL_NOTIFY_DEVICE WITH (VALUE_FORMAT='JSON') AS \
  SELECT CONNECT_TS + '    ' + USER_DEVICE_NAME + ' connected to ' + AP_NAME AS TEXT, \
         'device_connections' AS CHANNEL \
  FROM UBNT_AP_USER_DEVICE_CONNECTS5 \
  WHERE DEVICE_TYPE='Espressi';

  …on both desktop:

  

  …and mobile

  

  Recap

  We’ve come a long way! Let’s remind ourselves what we’ve built:

  

  1. Streaming syslog events from multiple devices into Kafka
  2. Streaming contents and subsequent changes from MongoDB into Kafka
  3. Filtering syslog events to just include those from Ubiquiti devices
  4. Enriching Ubiquiti syslog events with both (i) access point name and (ii) user device name and device type
  5. Streaming enriched Ubiquiti syslog data to Elasticsearch for analysis in Kibana
  6. Filtering Ubiquiti syslog data to include only wifi connections from devices which are wifi-plugs
  7. Triggering push notification from Kafka topic to Slack

  Phew!

  Summary

  In this mini-series of blog articles we’ve seen the power of Apache Kafka—including Kafka Connect—and KSQL to build applications and analytics on Apache Kafka as a streaming platform. Using KSQL, it’s simple to create scalable real-time stream processing applications, using just SQL—no other coding required! Kafka Connect provides a configuration-file based method for powerful streaming integration between sources of data into Kafka, and from Kafka out to targets such as Elasticsearch.

  If you’re interested in what KSQL can do, check out the rest of this:

  • Download Confluent Platform and follow the Quick Start to get started with KSQL
  • Watch the introductory video at the KSQL home page
  • Check out the clickstream demo that’s available as part of KSQL’s GitHub repo
  • Questions? Ask them in the #ksql channel on our community Slack group.

  The post We ❤️ syslogs: Real-time syslog processing with Apache Kafka and KSQL—Part 3: Enriching events with external data appeared first on Confluent.

  
  
  https://www.confluent.io/blog/real-time-syslog-processing-apache-kafka-ksql-enriching-events-with-external-data/
• 

  The Role of Service Partners with Confluent Cloud
  • sorangutan  

  1
  1
  Posts
  12
  Views

  

  Through Confluent Cloud’s service partners, our users can easily build scalable data streaming applications with fully managed Apache Kafka, connectors, security, and more.
  
  https://www.confluent.io/blog/benefits-of-confluent-clouds-service-partners/
• 

  The 5 Tenets of Data Integration
  • sorangutan  

  1
  1
  Posts
  16
  Views

  

  Data integration is the process of combining data into a single, unified view. Learn how it works, its benefits, use cases, and 5 principles for getting started.
  
  https://www.confluent.io/blog/data-integration-benefits-challenges-considerations/
• 

  Top 4 Reasons Confluent Needs to Be on Your AWS re:Invent 2022 Agenda
  • sorangutan  

  1
  1
  Posts
  19
  Views

  

  Learn to leverage Confluent for real-time business intelligence, build streaming data pipelines in minutes, meet with industry Kafka leaders, and end the night with DJ’s and dancing.
  
  https://www.confluent.io/blog/visit-confluent-at-aws-reinvent-2022/