Visualize Confluent information in Amazon QuickSight utilizing Amazon Athena

This can be a visitor put up written by Ahmed Saef Zamzam and Geetha Anne from Confluent.

Companies are utilizing real-time information streams to achieve insights into their firm’s efficiency and make knowledgeable, data-driven selections sooner. As real-time information has develop into important for companies, a rising variety of corporations are adapting their information technique to concentrate on information in movement. Occasion streaming is the central nervous system of a knowledge in movement technique and, in lots of organizations, Apache Kafka is the device that powers it.

At the moment, Kafka is well-known and broadly used for streaming information. Nevertheless, managing and working Kafka at scale can nonetheless be difficult. Confluent gives an answer via its totally managed, cloud-native service that simplifies operating and working information streams at scale. Confluent extends open-source Kafka via a collection of associated companies and options designed to reinforce the information in movement expertise for operators, builders, and designers in manufacturing.

On this put up, we exhibit how Amazon Athena, Amazon QuickSight, and Confluent work collectively to allow visualization of knowledge streams in near-real time. We use the Kafka connector in Athena to do the next:

• Be part of information inside Confluent with information saved in one of many many information sources supported by Athena, resembling Amazon Easy Storage Service (Amazon S3)
• Visualize Confluent information utilizing QuickSight

Challenges

Goal-built stream processing engines, like Confluent ksqlDB, typically present SQL-like semantics for real-time transformations, joins, aggregations, and filters on streaming information. With ksqlDB, you possibly can create persistent queries, which constantly course of streams of occasions in keeping with particular logic, and materialize streaming information in views that may be queried at a cut-off date (pull queries) or subscribed to by shoppers (push queries).

ksqlDB is one resolution that made stream processing accessible to a wider vary of customers. Nevertheless, pull queries, like these supported by ksqlDB, might not be appropriate for all stream processing use instances, and there could also be complexities or distinctive necessities that pull queries are usually not designed for.

Knowledge visualization for Confluent information

A frequent use case for enterprises is information visualization. To visualise information saved in Confluent, you should use one in all over 120 pre-built connectors, supplied by Confluent, to write down streaming information to a vacation spot information retailer of your selection. Subsequent, you join your small business intelligence (BI) device to the information retailer to start visualizing the information.

The next diagram depicts a typical structure utilized by many Confluent clients. On this workflow, information is written to Amazon S3 via the Confluent S3 sink connector after which analyzed with Athena, a serverless interactive analytics service that lets you analyze and question information saved in Amazon S3 and varied different information sources utilizing commonplace SQL. You may then use Athena as an enter information supply to QuickSight, a extremely scalable cloud native BI service, for additional evaluation.

Though this strategy works effectively for a lot of use instances, it requires information to be moved, and due to this fact duplicated, earlier than it may be visualized. This duplication not solely provides effort and time for information engineers who could must develop and check new scripts, but in addition creates information redundancy, making it tougher to handle and safe the information, and will increase storage value.

Enriching information with reference information in one other information retailer

With ksqlDB queries, the supply and vacation spot are all the time Kafka matters. Due to this fact, when you’ve got a knowledge stream that you might want to enrich with exterior reference information, you might have two choices. One possibility is to import the reference information into Confluent, mannequin it as a desk, and use ksqlDB’s stream-table be part of to counterpoint the stream. The opposite possibility is to ingest the information stream right into a separate information retailer and carry out be part of operations there. Each require information motion and lead to duplicate information storage.

Resolution overview

To date, now we have mentioned two challenges that aren’t addressed by typical stream processing instruments. Is there an answer that addresses each challenges concurrently?

Once you wish to analyze information with out separate pipelines and jobs, a well-liked selection is Athena. With Athena, you possibly can run SQL queries on a wide selection of knowledge sources—along with Amazon S3—with out studying a brand new language, creating scripts to extract (and duplicate) information, or managing infrastructure.

Just lately, Athena introduced a connector for Kafka. Like Athena’s different connectors, queries on Kafka are processed inside Kafka and return outcomes to Athena. The connector helps predicate pushdown, which implies that including filters to your queries can cut back the quantity of knowledge scanned, enhance question efficiency, and cut back value.

For instance, when utilizing this connector, the quantity of knowledge scanned by the question SELECT * FROM CONFLUENT_TABLE may very well be considerably larger than the quantity of knowledge scanned by the question SELECT * FROM CONFLUENT_TABLE WHERE COUNTRY = 'UK'. The reason being that the AWS Lambda perform which gives the runtime setting for the Athena connector, filters information on the supply earlier than returning it to Athena.

Let’s assume now we have a stream of on-line transactions flowing into Confluent and buyer reference information saved in Amazon S3. We wish to use Athena to hitch each information sources collectively and produce a brand new dataset for QuickSight. As a substitute of utilizing the S3 sink connector to load information into Amazon S3, we use Athena to question Confluent and be part of it with S3 information—all with out shifting information. The next diagram illustrates this structure.

We carry out the next steps:

1. Register the schema of your Confluent information.
2. Configure the Athena connector for Kafka.
3. Optionally, interactively analyze Confluent information.
4. Create a QuickSight dataset utilizing Athena because the supply.

Register the schema

To attach Athena to Confluent, the connector wants the schema of the subject to be registered within the AWS Glue Schema Registry, which Athena makes use of for question planning.

The next is a pattern report in Confluent:

{
  "transaction_id": "23e5ed25-5818-4d4f-acb3-73ef04d51d21",
  "customer_id": "126-58-9758",
  "quantity": 986,
  "timestamp": "2023-01-03T15:40:42",
  "product_category": "health_fitness"
}

The next is the schema of this report:

{
  "topicName": "transactions",
  "message": {
    "dataFormat": "json",
    "fields": [
      {
        "name": "transaction_id",
        "mapping": "transaction_id",
        "type": "VARCHAR"
      },
      {
        "name": "customer_id",
        "mapping": "customer_id",
        "type": "VARCHAR"
      },
      {
        "name": "amount",
        "mapping": "amount",
        "type": "INTEGER"
      },
      {
        "name": "timestamp",
        "mapping": "timestamp",
        "type": "timestamp",
        "formatHint": "yyyy-MM-dd'T'HH:mm:ss"
      },
      {
        "name": "product_category",
        "mapping": "product_category",
        "type": "VARCHAR"
      },
      {
        "name": "customer_id",
        "mapping": "customer_id",
        "type": "VARCHAR"
      }
    ]
  }
}

The information producer writing the information can register this schema with the AWS Glue Schema Registry. Alternatively, you should use the AWS Administration Console or AWS Command Line Interface (AWS CLI) to create a schema manually.

We create the schema manually by operating the next CLI command. Substitute <registry_name> together with your registry identify and be sure that the textual content within the description subject contains the required string {AthenaFederationKafka}:

aws glue create-registry –registry-name <registry_name> --description {AthenaFederationKafka}

Subsequent, we run the next command to create a schema contained in the newly created schema registry:

aws glue create-schema –registry-id RegistryName=<registry_name> --schema-name <schema_name> --compatibility <Compatibility_Mode> --data-format JSON –schema-definition <Schema>

Earlier than operating the command, you’ll want to present the next particulars:

• Substitute <registry_name> with our AWS Glue Schema Registry identify
• Substitute <schema_name> with the identify of our Confluent Cloud subject, for instance, transactions
• Substitute <Compatibility_Mode> with one of many supported compatibility modes, for instance, ‘Backward’
• Substitute <Schema> with our schema

Configure and deploy the Athena Connector

With our schema created, we’re able to deploy the Athena connector. Full the next steps:

1. On the Athena console, select Knowledge sources within the navigation pane.
2. Select Create information supply.
3. Seek for and choose Apache Kafka.
   
   
4. For Knowledge supply identify, enter the identify for the information supply.
   

This information supply identify might be referenced in your queries. For instance:

SELECT * 
FROM <data_source_name>.<registry_name>.<schema_name>
WHERE COL1='SOMETHING'

Making use of this to our use case and beforehand outlined schema, our question can be as follows:

SELECT * 
FROM "Confluent"."transactions_db"."transactions"
WHERE product_category='Children'

5. Within the Connection particulars part, select Create Lambda perform.
   

You’re redirected to the Purposes web page on the Lambda console. A number of the software settings are already stuffed.

The next are the essential settings required for integrating with Confluent Cloud. For extra data on these settings, check with Parameters.

6. For LambdaFunctionName, enter the identify for the Lambda perform the connector will use. For instance, athena_confluent_connector.

We use this parameter within the subsequent step.

7. For KafkaEndpoint, enter the Confluent Cloud bootstrap URL.

You’ll find this on the Cluster settings web page within the Confluent Cloud UI.

Confluent Cloud helps two authentication mechanisms: OAuth and SASL/PLAIN (API keys). The connector doesn’t help OAuth; this leaves us with SASL/PLAIN. SASL/PLAIN makes use of SSL as a safety protocol and PLAIN as SASL mechanism.

8. For AuthType, enter SASL_SSL_PLAIN.

The API key and secret utilized by the connector to entry Confluent have to be saved in AWS Secrets and techniques Supervisor.

9. Get your Confluent API key or create a brand new one.
10. Run the next AWS CLI command to create the key in Secrets and techniques Supervisor:

    aws secretsmanager create-secret 
        --name <SecretNamePrefix>
        --secret-string "{"username":"<Confluent_API_KEY>","password":"<Confluent_Secret>"}"

The key string ought to have two key-value pairs, one named username and the opposite password.

11. For SecretNamePrefix, enter the key identify prefix created within the earlier step.
12. If the Confluent cloud cluster is reachable over the web, depart SecurityGroupIds and SubnetIds clean. In any other case, your Lambda perform must run in a VPC that has connectivity to your Confluent Cloud community. Due to this fact, enter a safety group ID and three non-public subnet IDs on this VPC.
13. For SpillBucket, enter the identify of an S3 bucket the place the connector can spill information.

Athena connectors briefly retailer (spill) information to Amazon S3 for additional processing by Athena.

14. Choose I acknowledge that this app creates customized IAM roles and useful resource insurance policies.
15. Select Deploy.
16. Return to the Connection particulars part on the Athena console and for Lambda, enter the identify of the Lambda perform you created.
17. Select Subsequent.
    
18. Select Create information supply.

Carry out interactive evaluation on Confluent information

With the Athena connector arrange, our streaming information is now queryable from the identical service we use to research S3 information lakes. Subsequent, we use Athena to conduct point-in-time evaluation of transactions flowing via Confluent Cloud.

Aggregation

We will use commonplace SQL capabilities to mixture the information. For instance, we will get the income by product class:

SELECT product_category, SUM(quantity) AS Income
FROM "Confluent"."athena_blog"."transactions"
GROUP BY product_category
ORDER BY Income desc

Enrich transaction information with buyer information

The aggregation instance can be obtainable with ksqlDB pull queries. Nevertheless, Athena’s connector permits us to hitch the information with different information sources like Amazon S3.

In our use case, the transactions streamed to Confluent Cloud lack detailed details about clients, other than a customer_id. Nevertheless, now we have a reference dataset in Amazon S3 that has extra details about the shoppers. With Athena, we will be part of each datasets collectively to achieve insights about our clients. See the next code:

SELECT * 
FROM "Confluent"."athena_blog"."transactions" a
INNER JOIN "AwsDataCatalog"."athenablog"."buyer" b 
ON a.customer_id=b.customer_id

You may see from the outcomes that we have been capable of enrich the streaming information with buyer particulars, saved in Amazon S3, together with identify and deal with.

Visualize information utilizing QuickSight

One other highly effective function this connector brings is the power to visualise information saved in Confluent utilizing any BI device that helps Athena as a knowledge supply. On this put up, we use QuickSight. QuickSight is a machine studying (ML)-powered BI service constructed for the cloud. You should use it to ship easy-to-understand insights to the individuals you’re employed with, wherever they’re.

For extra details about signing up for QuickSight, see Signing up for an Amazon QuickSight subscription.

Full the next steps to visualise your streaming information with QuickSight:

1. On the QuickSight console, select Datasets within the navigation pane.
2. Select New dataset.
3. Select Athena as the information supply.
4. For Knowledge supply identify, enter a reputation.
5. Select Create information supply.
6. Within the Select your desk part, select Use customized SQL.
   
7. Enter the be part of question just like the one given beforehand, then select Affirm question.
   
8. Subsequent, select to import the information into SPICE (Tremendous-fast, Parallel, In-memory Calculation Engine), a totally managed in-memory cache that reinforces efficiency, or straight question the information.

Using SPICE will improve efficiency, however the information could have to be periodically up to date. You may select to incrementally refresh your dataset or schedule common refreshes with SPICE. If you need near-real-time information mirrored in your dashboards, choose Instantly question your information. Be aware that with the direct question possibility, person actions in QuickSight, resembling making use of a drill-down filter, could invoke a brand new Athena question.

9. Select Visualize.
   

That’s it, now we have efficiently linked QuickSight to Confluent via Athena. With only a few clicks, you possibly can create a number of visuals displaying information from Confluent.

Clear up

To keep away from incurring ongoing costs, delete the sources you provisioned by finishing the next steps:

1. Delete the AWS Glue schema and registry.
2. Delete the Athena Kafka connector.
3. Delete the QuickSight dataset.

Conclusion

On this put up, we mentioned use instances for Athena and Confluent. We supplied examples of how you should use each for near-real-time information visualization with QuickSight and interactive evaluation involving joins between streaming information in Confluent and information saved in Amazon S3.

The Athena connector for Kafka simplifies the method of querying and analyzing streaming information from Confluent Cloud. It removes the necessity to first transfer streaming information to persistent storage earlier than it may be utilized in downstream use instances like enterprise intelligence. This enhances the present integration between Confluent and Athena, utilizing the S3 sink connector, which allows loading streaming information into a knowledge lake, and is an extra possibility for purchasers who wish to allow interactive evaluation on Confluent information.

In regards to the authors

Ahmed Zamzam is a Senior Associate Options Architect at Confluent, with a concentrate on the AWS partnership. In his function, he works with clients within the EMEA area throughout varied industries to help them in constructing functions that leverage their information utilizing Confluent and AWS. Previous to Confluent, Ahmed was a Specialist Options Architect for Analytics AWS specialised in information streaming and search. In his free time, Ahmed enjoys touring, taking part in tennis, and biking.

Geetha Anne is a Associate Options Engineer at Confluent with earlier expertise in implementing options for data-driven enterprise issues on the cloud, involving information warehousing and real-time streaming analytics. She fell in love with distributed computing throughout her undergraduate days and has adopted her curiosity ever since. Geetha gives technical steering, design recommendation, and thought management to key Confluent clients and companions. She additionally enjoys educating complicated technical ideas to each tech-savvy and basic audiences.