Posts Tagged Data Lake

Utilizing In-memory Data Caching to Enhance the Performance of Data Lake-based Applications

Posted by Gary A. Stafford in Analytics, AWS, Big Data, Cloud, Java Development, Software Development, SQL on July 12, 2022

Significantly improve the performance and reduce the cost of data lake-based analytics applications using Amazon ElastiCache for Redis

Introduction

The recent post, Developing Spring Boot Applications for Querying Data Lakes on AWS using Amazon Athena, demonstrated how to develop a Cloud-native analytics application using Spring Boot. The application queried data in an Amazon S3-based data lake via an AWS Glue Data Catalog utilizing the Amazon Athena API.

Securely exposing data in a data lake using RESTful APIs can fulfill many data-consumer needs. However, access to that data can be significantly slower than access from a database or data warehouse. For example, in the previous post, we imported the OpenAPI v3 specification from the Spring Boot service into Postman. The API specification contained approximately 17 endpoints.

Running a suite of integration tests against the Spring Boot service using Postman

From my local development laptop, the Postman API test run times for all service endpoints took an average of 32.4 seconds. The Spring Boot service was running three Kubernetes pod replicas on Amazon EKS in the AWS US East (N. Virginia) Region.

Sample of test results ran from service (no Redis cache)

Compare the data lake query result times to equivalent queries made against a minimally-sized Amazon RDS for PostgreSQL database instance containing the same data. The average run times for all PostgreSQL queries averaged 10.8 seconds from a similar Spring Boot service. Although not a precise benchmark, we can clearly see that access to the data in the Amazon S3-based data lake is substantially slower, approximately 3x slower, than that of the PostgreSQL database. Tuning the database would easily create an even greater disparity.

Sample of test results for queries ran from service against Athena vs. PostgreSQL (quicker is better)

Caching for Data Lakes

According to AWS, the speed and throughput of a database can be the most impactful factor for overall application performance. Consequently, in-memory data caching can be one of the most effective strategies to improve overall application performance and reduce database costs. This same caching strategy can be applied to analytics applications built atop data lakes, as this post will demonstrate.

High-level AWS architecture demonstrated in this post

In-memory Caching

According to Hazelcast, memory caching (aka in-memory caching), often simply referred to as caching, is a technique in which computer applications temporarily store data in a computer’s main memory (e.g., RAM) to enable fast retrievals of that data. The RAM used for the temporary storage is known as the cache. As an application tries to read data, typically from a data storage system like a database, it checks to see if the desired record already exists in the cache. If it does, the application will read the data from the cache, thus eliminating the slower access to the database. If the desired record is not in the cache, then the application reads the record from the source. When it retrieves that data, it writes it to the cache so that when the application needs that same data in the future, it can quickly retrieve it from the cache.

Further, according to Hazelcast, as an application tries to read data, typically from a data storage system like a database, it checks to see if the desired record already exists in the cache. If it does, the application will read the data from the cache, thus eliminating the slower access to the database. If the desired record is not in the cache, then the application reads the record from the source. When it retrieves that data, it writes it to the cache so that when the application needs that same data in the future, it can quickly retrieve it from the cache.

Redis In-memory Data Store

According to their website, Redis is the open-source, in-memory data store used by millions of developers as a database, cache, streaming engine, and message broker. Redis provides data structures such as strings, hashes, lists, sets, sorted sets with range queries, bitmaps, hyperloglogs, geospatial indexes, and streams. In addition, Redis has built-in replication, Lua scripting, LRU eviction, transactions, and different levels of on-disk persistence and provides high availability via Redis Sentinel and automatic partitioning with Redis Cluster.

Amazon ElastiCache for Redis

According to AWS, Amazon ElastiCache for Redis, the fully-managed version of Redis, according to AWS, is a blazing fast in-memory data store that provides sub-millisecond latency to power internet-scale real-time applications. Redis applications can work seamlessly with ElastiCache for Redis without any code changes. ElastiCache for Redis combines the speed, simplicity, and versatility of open-source Redis with manageability, security, and scalability from AWS. As a result, Redis is an excellent choice for implementing a highly available in-memory cache to decrease data access latency, increase throughput, and ease the load on relational and NoSQL databases.

ElastiCache Performance Results

In the following post, we will add in-memory caching to the Spring Boot service introduced in the previous post. In preliminary tests with Amazon ElastiCache for Redis, the Spring Boot service delivered a 34x improvement in average response times. For example, test runs with the best-case scenario of a Redis cache hit ratio of 100% averaged 0.95 seconds compared to 32.4 seconds without Redis.

Sample of uncached versus cached test results (quicker is better)

Source Code

All the source code and Docker and Kubernetes resources are open-source and available on GitHub.

git clone --depth 1 -b redis \
    https://github.com/garystafford/athena-spring-app.git

In addition, a Docker image for the Redis-base Spring Boot service is available on Docker Hub. For this post, use the latest tag with the .redis suffix.

Spring Boot service images are available on Docker Hub

Code Changes

The following code changes are required to the Spring Boot service to implement Spring Boot Cache with Redis.

Gradle Build

The gradle.build file now implements two additional dependencies, Spring Boot’s spring-boot-starter-cache and spring-boot-starter-data-redis (lines 45–46).

	plugins {
	id 'org.springframework.boot' version '2.7.1'
	id 'io.spring.dependency-management' version '1.0.11.RELEASE'
	id 'java'
	id 'io.freefair.lombok' version '6.5.0-rc1'
	}

	group = 'aws.example'
	version = '2.0.0'
	sourceCompatibility = '17'

	def awsSdkVersion = '2.17.225'
	def springBootVersion = '2.7.1'
	def restAssuredVersion = '5.1.1'

	repositories {
	mavenCentral()
	}

	dependencies {
	// aws sdk
	runtimeOnly "software.amazon.awssdk:bom:${awsSdkVersion}"
	implementation "software.amazon.awssdk:athena:${awsSdkVersion}"

	// spring
	annotationProcessor "org.springframework.boot:spring-boot-configuration-processor:${springBootVersion}"
	implementation "org.springframework.boot:spring-boot-starter-web:${springBootVersion}"
	implementation "org.springframework.boot:spring-boot-starter-web:${springBootVersion}"
	implementation "org.springframework.boot:spring-boot-starter-actuator:${springBootVersion}"
	developmentOnly "org.springframework.boot:spring-boot-devtools:${springBootVersion}"
	implementation 'org.springdoc:springdoc-openapi-ui:1.6.9'
	implementation 'org.springframework:spring-context:5.3.20'

	// testings
	testImplementation "org.springframework.boot:spring-boot-starter-test:${springBootVersion}"
	testImplementation "io.rest-assured:rest-assured:${restAssuredVersion}"
	testImplementation "io.rest-assured:json-path:${restAssuredVersion}"
	testImplementation "io.rest-assured:xml-path:${restAssuredVersion}"
	testImplementation "io.rest-assured:json-schema-validator:${restAssuredVersion}"

	// monitoring
	implementation 'io.micrometer:micrometer-registry-prometheus:1.9.1'

	// spring caching with redis
	implementation "org.springframework.boot:spring-boot-starter-cache:${springBootVersion}"
	implementation "org.springframework.boot:spring-boot-starter-data-redis:${springBootVersion}"
	}

	tasks.named('test') {
	useJUnitPlatform()
	}

view raw build.gradle hosted with ❤ by GitHub

Application Properties

The application properties file, application.yml, has been modified for both the dev and prod Spring Profiles. The dev Spring Profile expects Redis to be running on localhost. Correspondingly, the project’s docker-compose.yml file now includes a Redis container for local development. The time-to-live (TTL) for all Redis caches is arbitrarily set to one minute for dev and five minutes for prod. To increase application performance and reduce the cost of querying the data lake using Athena, increase Redis’s TTL. Note that increasing the TTL will reduce data freshness.

	spring:
	profiles:
	active: dev
	server:
	port: 8080
	servlet:
	contextPath: /v1
	athena:
	region: us-east-1
	workgroup: primary
	catalog: AwsDataCatalog
	database: tickit_demo
	limit: 25
	client-execution-timeout: 100000
	retry-sleep: 1000
	results-bucket: ${RESULTS_BUCKET}
	named-query-id: ${NAMED_QUERY_ID}
	—
	spring:
	config:
	activate:
	on-profile: dev
	redis:
	host: localhost
	port: 6379
	cache:
	redis:
	time-to-live: 60000 # 1000 ms * 60 * 1 = 1 min
	enable-statistics: true
	server:
	port: 8080
	logging:
	level:
	root: DEBUG
	management:
	endpoints:
	web:
	exposure:
	include: '*'
	jmx:
	exposure:
	include: '*'
	—
	spring:
	config:
	activate:
	on-profile: prod
	redis:
	host: ${REDIS_HOST}
	port: 6379
	cache:
	redis:
	time-to-live: 300000 # 1000 ms * 60 * 5 = 5 min
	enable-statistics: true
	logging:
	level:
	root: INFO
	management:
	endpoints:
	web:
	exposure:
	include: health, prometheus
	jmx:
	exposure:
	include: health

view raw application.yml hosted with ❤ by GitHub

Athena Application Class

The AthenaApplication class declaration is now decorated with Spring Framework’s EnableCaching annotation (line 22). Additionally, two new Beans have been added (lines 58–68). Spring Redis provides an implementation for the Spring cache abstraction through the org.springframework.data.redis.cache package. The RedisCacheManager cache manager creates caches by default upon the first write. The RedisCacheConfiguration cache configuration helps to customize RedisCache behavior such as caching null values, cache key prefixes, and binary serialization.

	package com.example.athena;

	import com.example.athena.common.PreparedStatement;
	import com.example.athena.common.View;
	import com.example.athena.config.ConfigProperties;
	import org.springframework.beans.factory.annotation.Autowired;
	import org.springframework.boot.SpringApplication;
	import org.springframework.boot.autoconfigure.SpringBootApplication;
	import org.springframework.boot.context.properties.EnableConfigurationProperties;
	import org.springframework.cache.annotation.EnableCaching;
	import org.springframework.context.annotation.Bean;
	import org.springframework.data.redis.cache.RedisCacheConfiguration;
	import org.springframework.data.redis.cache.RedisCacheManager;
	import org.springframework.data.redis.connection.RedisConnectionFactory;
	import org.springframework.web.servlet.config.annotation.CorsRegistry;
	import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;

	import java.time.Duration;

	@SpringBootApplication
	@EnableConfigurationProperties(ConfigProperties.class)
	@EnableCaching
	public class AthenaApplication {
	private final PreparedStatement preparedStatement;

	private final View view;

	@Autowired
	public AthenaApplication(PreparedStatement preparedStatement, View view) {
	this.preparedStatement = preparedStatement;
	this.view = view;
	}

	public static void main(String[] args) {
	SpringApplication.run(AthenaApplication.class, args);
	}

	@Bean
	void CreatePreparedStatement() {
	preparedStatement.CreatePreparedStatement();
	}

	@Bean
	void createView() {
	view.CreateView();
	}

	@Bean
	public WebMvcConfigurer corsConfigurer() {
	return new WebMvcConfigurer() {
	@Override
	public void addCorsMappings(CorsRegistry registry) {
	registry.addMapping("/*").allowedOrigins("");
	}
	};
	}

	@Bean
	public RedisCacheManager cacheManager(RedisConnectionFactory connectionFactory) {
	return RedisCacheManager.create(connectionFactory);
	}

	@Bean
	public RedisCacheConfiguration cacheConfiguration() {
	return RedisCacheConfiguration.defaultCacheConfig()
	.entryTtl(Duration.ofMinutes(5))
	.disableCachingNullValues();
	}

	}

view raw AthenaApplication.java hosted with ❤ by GitHub

POJO Data Model Classes

Spring Boot Redis caching uses Java serialization and deserialization. Therefore, all the POJO data model classes must implement Serializable (line 14).

	package com.example.athena.tickit.model.ecomm;

	import lombok.AllArgsConstructor;
	import lombok.Data;
	import lombok.NoArgsConstructor;

	import java.io.Serializable;
	import java.math.BigDecimal;
	import java.time.LocalDateTime;

	@Data
	@NoArgsConstructor
	@AllArgsConstructor
	public class Listing implements Serializable {

	private int id;

	private int sellerId;

	private int eventId;

	private int dateId;

	private int numTickets;

	private BigDecimal pricePerTicket;

	private LocalDateTime listTime;

	}

view raw Listing.java hosted with ❤ by GitHub

Service Classes

Each public method in the Service classes is now decorated with Spring Framework’s Cachable annotation (lines 42 and 66). For example, the findById(int id) method in the CategoryServiceImp class is annotated with @Cacheable(value = "categories", key = "#id"). The method’s key parameter uses Spring Expression Language (SpEL) expression for computing the key dynamically. Default is null, meaning all method parameters are considered as a key unless a custom keyGenerator has been configured. If no value is found in the Redis cache for the computed key, the target method will be invoked, and the returned value will be stored in the associated cache.

	package com.example.athena.tickit.service;

	import com.example.athena.common.AthenaClientFactory;
	import com.example.athena.common.AthenaCommon;
	import com.example.athena.config.ConfigProperties;
	import com.example.athena.tickit.model.ecomm.Listing;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;
	import org.springframework.beans.factory.annotation.Autowired;
	import org.springframework.cache.annotation.Cacheable;
	import org.springframework.stereotype.Service;
	import software.amazon.awssdk.services.athena.AthenaClient;
	import software.amazon.awssdk.services.athena.model.*;
	import software.amazon.awssdk.services.athena.paginators.GetQueryResultsIterable;

	import java.math.BigDecimal;
	import java.time.LocalDateTime;
	import java.time.format.DateTimeFormatter;
	import java.util.ArrayList;
	import java.util.List;

	import static java.lang.Integer.parseInt;

	@Service
	public class ListingServiceImp implements ListingService {

	private static final Logger logger = LoggerFactory.getLogger(ListingServiceImp.class);

	private final ConfigProperties configProperties;

	private final AthenaClientFactory athenaClientFactory;

	private final AthenaCommon athenaCommon;

	@Autowired
	public ListingServiceImp(ConfigProperties configProperties, AthenaClientFactory athenaClientFactory, AthenaCommon athenaCommon) {
	this.configProperties = configProperties;
	this.athenaClientFactory = athenaClientFactory;
	this.athenaCommon = athenaCommon;
	}

	@Cacheable(value = "listings")
	public List<Listing> findAll(Integer limit, Integer offset) {

	if (limit == null \|\| limit < 1 \|\| limit > configProperties.getLimit()) {
	limit = configProperties.getLimit();
	}

	if (offset == null \|\| offset < 1) {
	offset = 0;
	}

	String whereClause = "WHERE listid IS NOT NULL";

	String query = String.format("""
	SELECT *
	FROM refined_tickit_public_listing
	%s
	ORDER BY listid
	OFFSET %s
	LIMIT %s;""", whereClause, offset, limit);

	return startQuery(query);
	}

	@Cacheable(value = "listing", key = "#id")
	public Listing findById(int id) {
	String query = String.format("""
	SELECT DISTINCT *
	FROM refined_tickit_public_listing
	WHERE listid=%s""", id);

	Listing listing;
	try {
	listing = startQuery(query).get(0);
	} catch (java.lang.IndexOutOfBoundsException e) {
	logger.error(e.getMessage());
	return null;
	}
	return listing;
	}

	private List<Listing> startQuery(String query) {

	logger.debug(String.format("Query: %s", query.replace("\n", " ")));

	AthenaClient athenaClient = athenaClientFactory.createClient();
	String queryExecutionId = athenaCommon.submitAthenaQuery(athenaClient, query);
	athenaCommon.waitForQueryToComplete(athenaClient, queryExecutionId);
	List<Listing> listings = processResultRows(athenaClient, queryExecutionId);
	athenaClient.close();

	return listings;
	}

	private List<Listing> processResultRows(AthenaClient athenaClient, String queryExecutionId) {

	List<Listing> listings = new ArrayList<>();

	try {
	GetQueryResultsRequest getQueryResultsRequest = GetQueryResultsRequest.builder()
	.queryExecutionId(queryExecutionId).build();

	GetQueryResultsIterable getQueryResultsResults = athenaClient.getQueryResultsPaginator(getQueryResultsRequest);

	List<Row> rows;

	DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.S");

	for (GetQueryResultsResponse result : getQueryResultsResults) {
	rows = result.resultSet().rows();

	for (Row myRow : rows.subList(1, rows.size())) { // skip first row – column names
	List<Datum> allData = myRow.data();
	Listing listing = new Listing();
	listing.setId(parseInt(allData.get(0).varCharValue()));
	listing.setSellerId(parseInt(allData.get(1).varCharValue()));
	listing.setEventId(parseInt(allData.get(2).varCharValue()));
	listing.setDateId(parseInt(allData.get(3).varCharValue()));
	listing.setNumTickets(Integer.parseInt(allData.get(4).varCharValue()));
	listing.setPricePerTicket(new BigDecimal(allData.get(5).varCharValue()));
	listing.setListTime(LocalDateTime.parse(allData.get(6).varCharValue(), formatter));
	listings.add(listing);
	}
	}
	} catch (AthenaException e) {
	logger.error(e.getMessage());
	}
	return listings;
	}
	}

view raw ListingServiceImp.java hosted with ❤ by GitHub

Controller Classes

There are no changes required to the Controller classes.

Amazon ElastiCache for Redis

Multiple options are available for creating an Amazon ElastiCache for Redis cluster, including cluster mode, multi-AZ option, auto-failover option, node type, number of replicas, number of shards, replicas per shard, Availability Zone placements, and encryption at rest and encryption in transit options. The results in this post are based on a minimally-configured Redis version 6.2.6 cluster, with one shard, two cache.r6g.large nodes, and cluster mode, multi-AZ option, and auto-failover all disabled. In addition, encryption at rest and encryption in transit were also disabled. This cluster configuration is sufficient for development and testing, but not Production.

Amazon ElastiCache for Redis cluster monitoring console showing caching activity

Testing the Cache

To test Amazon ElastiCache for Redis, we will use Postman again with the imported OpenAPI v3 specification. With all data evicted from existing Redis caches, the first time the Postman tests run, they cause the service’s target methods to be invoked and the returned data stored in the associated caches.

To confirm this caching behavior, use the Redis CLI’s --scan option. To access the redis-cli, I deployed a single Redis pod to Amazon EKS. The first time the --scan command runs, we should get back an empty list of keys. After the first Postman test run, the same --scan command should return a list of cached keys.

Use the Redis CLI’s MONITOR option to further confirm data is being cached, as indicated by the set command.

Athena query results being cached in Redis

After the initial caching of data, use the Redis CLI’s MONITOR option, again, to confirm the cache is being hit instead of calling the target methods, which would then invoke the Athena API. Rerunning the Postman tests, we should see get commands as opposed to set commands.

Lastly, to confirm the Spring Boot service is effectively using Redis to cache data, we can also check Amazon Athena’s Recent queries tab in the AWS Management Console. After repeated sequential test runs within the TTL window, we should only see one Athena query per endpoint.

Amazon Athena Recent queries tab in the AWS Management Console

Conclusion

In this brief follow-up to the recent post, Developing Spring Boot Applications for Querying Data Lakes on AWS using Amazon Athena, we saw how to substantially increase data lake application performance using Amazon ElastiCache for Redis. Although this caching technique is often associated with databases, it can also be effectively applied to data lake-based applications, as demonstrated in the post.

This blog represents my viewpoints and not of my employer, Amazon Web Services (AWS). All product names, logos, and brands are the property of their respective owners. All diagrams and illustrations are property of the author unless otherwise noted.

Amazon Athena, AWS, Data Lake, Java, Redis, Spring Boot

Leave a comment

Developing Spring Boot Applications for Querying Data Lakes on AWS using Amazon Athena

Posted by Gary A. Stafford in Analytics, AWS, Big Data, Cloud, Java Development, Software Development on June 26, 2022

Learn how to develop Cloud-native, RESTful Java services that query data in an AWS-based data lake using Amazon Athena’s API

Introduction

AWS provides a collection of fully-managed services that makes building and managing secure data lakes faster and easier, including AWS Lake Formation, AWS Glue, and Amazon S3. Additional analytics services such as Amazon EMR, AWS Glue Studio, and Amazon Redshift allow Data Scientists and Analysts to run high-performance queries on large volumes of semi-structured and structured data quickly and economically.

What is not always as obvious is how teams develop internal and external customer-facing analytics applications built on top of data lakes. For example, imagine sellers on an eCommerce platform, the scenario used in this post, want to make better marketing decisions regarding their products by analyzing sales trends and buyer preferences. Further, suppose the data required for the analysis must be aggregated from multiple systems and data sources; the ideal use case for a data lake.

Example of a personalized sales report generated from the Spring Boot service’s `salesbyseller` endpoint

In this post, we will explore an example Java Spring Boot RESTful Web Service that allows end-users to query data stored in a data lake on AWS. The RESTful Web Service will access data stored as Apache Parquet in Amazon S3 through an AWS Glue Data Catalog using Amazon Athena. The service will use Spring Boot and the AWS SDK for Java to expose a secure, RESTful Application Programming Interface (API).

Amazon Athena is a serverless, interactive query service based on Presto, used to query data and analyze big data in Amazon S3 using standard SQL. Using Athena functionality exposed by the AWS SDK for Java and Athena API, the Spring Boot service will demonstrate how to access tables, views, prepared statements, and saved queries (aka named queries).

TL;DR

Do you want to explore the source code for this post’s Spring Boot service or deploy it to Kubernetes before reading the full article? All the source code, Docker, and Kubernetes resources are open-source and available on GitHub.

git clone --depth 1 -b main \
    https://github.com/garystafford/athena-spring-app.git

A Docker image for the Spring Boot service is also available on Docker Hub.

Spring Boot service image available on Docker Hub

Data Lake Data Source

There are endless data sources to build a demonstration data lake on AWS. This post uses the TICKIT sample database provided by AWS and designed for Amazon Redshift, AWS’s cloud data warehousing service. The database consists of seven tables. Two previous posts and associated videos, Building a Data Lake on AWS with Apache Airflow and Building a Data Lake on AWS, detail the setup of the data lake used in this post using AWS Glue and optionally Apache Airflow with Amazon MWAA.

Those two posts use the data lake pattern of segmenting data as bronze (aka raw), silver (aka refined), and gold (aka aggregated), popularized by Databricks. The data lake simulates a typical scenario where data originates from multiple sources, including an e-commerce platform, a CRM system, and a SaaS provider must be aggregated and analyzed.

High-level data lake architecture demonstrated in the previous post

Spring Projects with IntelliJ IDE

Although not a requirement, I used JetBrains IntelliJ IDEA 2022 (Ultimate Edition) to develop and test the post’s Spring Boot service. Bootstrapping Spring projects with IntelliJ is easy. Developers can quickly create a Spring project using the Spring Initializr plugin bundled with the IntelliJ.

JetBrains IntelliJ IDEA plugin support for Spring projects

The Spring Initializr plugin’s new project creation wizard is based on start.spring.io. The plugin allows you to quickly select the Spring dependencies you want to incorporate into your project.

Adding dependencies to a new Spring project in IntelliJ

Visual Studio Code

There are also several Spring extensions for the popular Visual Studio Code IDE, including Microsoft’s Spring Initializr Java Support extension.

Gradle

This post uses Gradle instead of Maven to develop, test, build, package, and deploy the Spring service. Based on the packages selected in the new project setup shown above, the Spring Initializr plugin’s new project creation wizard creates a build.gradle file. Additional packages, such as Lombak, Micrometer, and Rest Assured, were added separately.

	plugins {
	id 'org.springframework.boot' version '2.7.1'
	id 'io.spring.dependency-management' version '1.0.11.RELEASE'
	id 'java'
	id 'io.freefair.lombok' version '6.5.0-rc1'
	}

	group = 'aws.example'
	version = '1.0.0'
	sourceCompatibility = '17'

	def awsSdkVersion = '2.17.225'
	def springBootVersion = '2.7.1'
	def restAssuredVersion = '5.1.1'

	repositories {
	mavenCentral()
	}

	dependencies {
	// aws sdk
	runtimeOnly "software.amazon.awssdk:bom:${awsSdkVersion}"
	implementation "software.amazon.awssdk:athena:${awsSdkVersion}"

	// spring
	annotationProcessor "org.springframework.boot:spring-boot-configuration-processor:${springBootVersion}"
	implementation "org.springframework.boot:spring-boot-starter-web:${springBootVersion}"
	implementation "org.springframework.boot:spring-boot-starter-web:${springBootVersion}"
	implementation "org.springframework.boot:spring-boot-starter-actuator:${springBootVersion}"
	developmentOnly "org.springframework.boot:spring-boot-devtools:${springBootVersion}"
	implementation 'org.springdoc:springdoc-openapi-ui:1.6.9'
	implementation 'org.springframework:spring-context:5.3.20'

	// testings
	testImplementation "org.springframework.boot:spring-boot-starter-test:${springBootVersion}"
	testImplementation "io.rest-assured:rest-assured:${restAssuredVersion}"
	testImplementation "io.rest-assured:json-path:${restAssuredVersion}"
	testImplementation "io.rest-assured:xml-path:${restAssuredVersion}"
	testImplementation "io.rest-assured:json-schema-validator:${restAssuredVersion}"

	// monitoring
	implementation 'io.micrometer:micrometer-registry-prometheus:1.9.1'
	}

	tasks.named('test') {
	useJUnitPlatform()
	}

view raw build.gradle hosted with ❤ by GitHub

Amazon Corretto

The Spring boot service is developed for and compiled with the most recent version of Amazon Corretto 17. According to AWS, Amazon Corretto is a no-cost, multiplatform, production-ready distribution of the Open Java Development Kit (OpenJDK). Corretto comes with long-term support that includes performance enhancements and security fixes. Corretto is certified as compatible with the Java SE standard and is used internally at Amazon for many production services.

Source Code

Each API endpoint in the Spring Boot RESTful Web Service has a corresponding POJO data model class, service interface and service implementation class, and controller class. In addition, there are also common classes such as configuration, a client factory, and Athena-specific request/response methods. Lastly, there are additional class dependencies for views and prepared statements.

Java class relationships related to querying the Amazon Athena `refined_tickit_public_category` table

The project’s source code is arranged in a logical hierarchy by package and class type.

	.
	└── com
	└── example
	└── athena
	├── AthenaApplication.java
	├── common
	│ ├── AthenaClientFactory.java
	│ ├── AthenaClientFactoryImp.java
	│ ├── AthenaCommon.java
	│ ├── NamedQuery.java
	│ ├── PreparedStatement.java
	│ └── View.java
	├── config
	│ └── ConfigProperties.java
	└── tickit
	├── controller
	│ ├── BuyerLikesByCategoryController.java
	│ ├── CategoryController.java
	│ ├── DateDetailController.java
	│ ├── EventController.java
	│ ├── ListingController.java
	│ ├── SaleBySellerController.java
	│ ├── SaleController.java
	│ ├── SalesByCategoryController.java
	│ ├── UserController.java
	│ └── VenueController.java
	├── model
	│ ├── crm
	│ │ └── User.java
	│ ├── ecomm
	│ │ ├── DateDetail.java
	│ │ ├── Listing.java
	│ │ └── Sale.java
	│ ├── resultsets
	│ │ ├── BuyerLikesByCategory.java
	│ │ ├── SaleBySeller.java
	│ │ └── SalesByCategory.java
	│ └── saas
	│ ├── Category.java
	│ ├── Event.java
	│ └── Venue.java
	└── service
	├── BuyerLikesByCategoryServiceImp.java
	├── BuyersLikesByCategoryService.java
	├── CategoryService.java
	├── CategoryServiceImp.java
	├── DateDetailService.java
	├── DateDetailServiceImp.java
	├── EventService.java
	├── EventServiceImp.java
	├── ListingService.java
	├── ListingServiceImp.java
	├── SaleBySellerService.java
	├── SaleBySellerServiceImp.java
	├── SaleService.java
	├── SaleServiceImp.java
	├── SalesByCategoryService.java
	├── SalesByCategoryServiceImp.java
	├── UserService.java
	├── UserServiceImp.java
	├── VenueService.java
	└── VenueServiceImp.java

view raw athena_spring_project_structure.txt hosted with ❤ by GitHub

Amazon Athena Access

There are three standard methods for accessing Amazon Athena with the AWS SDK for Java: 1) the AthenaClient service client, 2) the AthenaAsyncClient service client for accessing Athena asynchronously, and 3) using the JDBC driver with the AWS SDK. The AthenaClient and AthenaAsyncClient service clients are both parts of the software.amazon.awssdk.services.athena package. For simplicity, this post’s Spring Boot service uses the AthenaClient service client instead of Java’s asynchronously programming model. AWS supplies basic code samples as part of their documentation as a starting point for writing Athena applications using the SDK. The code samples also use the AthenaClient service client.

POJO-based Data Model Class

For each API endpoint in the Spring Boot RESTful Web Service, there is a corresponding Plain Old Java Object (POJO). According to Wikipedia, a POGO is an ordinary Java object, not bound by any particular restriction. The POJO class is similar to a JPA Entity, representing persistent data stored in a relational database. In this case, the POJO uses Lombok’s @Data annotation. According to the documentation, this annotation generates getters for all fields, a useful toString method, and hashCode and equals implementations that check all non-transient fields. It also generates setters for all non-final fields and a constructor.

	package com.example.athena.tickit.model.saas;

	import lombok.AllArgsConstructor;
	import lombok.Data;
	import lombok.NoArgsConstructor;

	import java.time.LocalDateTime;

	@Data
	@NoArgsConstructor
	@AllArgsConstructor
	public class Event {

	private int id;

	private int venueId;

	private int catId;

	private int dateId;

	private String name;

	private LocalDateTime startTime;

	}

view raw Event.java hosted with ❤ by GitHub

Each POJO corresponds directly to a ‘silver’ table in the AWS Glue Data Catalog. For example, the Event POJO corresponds to the refined_tickit_public_event table in the tickit_demo Data Catalog database. The POJO defines the Spring Boot service’s data model for data read from the corresponding AWS Glue Data Catalog table.

Glue Data Catalog refined_tickit_public_event table

The Glue Data Catalog table is the interface between the Athena query and the underlying data stored in Amazon S3 object storage. The Athena query targets the table, which returns the underlying data from S3.

Tickit Category data stored as Apache Parquet files in Amazon S3

Service Class

Retrieving data from the data lake via AWS Glue, using Athena, is handled by a service class. For each API endpoint in the Spring Boot RESTful Web Service, there is a corresponding Service Interface and implementation class. The service implementation class uses Spring Framework’s @Service annotation. According to the documentation, it indicates that an annotated class is a “Service,” initially defined by Domain-Driven Design (Evans, 2003) as “an operation offered as an interface that stands alone in the model, with no encapsulated state.” Most importantly for the Spring Boot service, this annotation serves as a specialization of @Component, allowing for implementation classes to be autodetected through classpath scanning.

	package com.example.athena.tickit.service;

	import com.example.athena.common.AthenaClientFactory;
	import com.example.athena.common.AthenaCommon;
	import com.example.athena.config.ConfigProperties;
	import com.example.athena.tickit.model.saas.Event;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;
	import org.springframework.beans.factory.annotation.Autowired;
	import org.springframework.stereotype.Service;
	import software.amazon.awssdk.services.athena.AthenaClient;
	import software.amazon.awssdk.services.athena.model.*;
	import software.amazon.awssdk.services.athena.paginators.GetQueryResultsIterable;

	import java.time.LocalDateTime;
	import java.time.format.DateTimeFormatter;
	import java.util.ArrayList;
	import java.util.List;

	import static java.lang.Integer.parseInt;

	@Service
	public class EventServiceImp implements EventService {

	private static final Logger logger = LoggerFactory.getLogger(EventServiceImp.class);

	private final ConfigProperties configProperties;

	private final AthenaClientFactory athenaClientFactory;

	private final AthenaCommon athenaCommon;

	@Autowired
	public EventServiceImp(ConfigProperties configProperties, AthenaClientFactory athenaClientFactory, AthenaCommon athenaCommon) {
	this.configProperties = configProperties;
	this.athenaClientFactory = athenaClientFactory;
	this.athenaCommon = athenaCommon;
	}

	public List<Event> findAll(Integer limit, Integer offset) {

	if (limit == null \|\| limit < 1 \|\| limit > configProperties.getLimit()) {
	limit = configProperties.getLimit();
	}

	if (offset == null \|\| offset < 1) {
	offset = 0;
	}

	String whereClause = "WHERE eventid IS NOT NULL";

	String query = String.format("""
	SELECT *
	FROM refined_tickit_public_event
	%s
	ORDER BY eventid
	OFFSET %s
	LIMIT %s;""", whereClause, offset, limit);

	return startQuery(query);
	}

	public Event findById(int id) {
	String query = String.format("""
	SELECT DISTINCT *
	FROM refined_tickit_public_event
	WHERE eventid=%s""", id);

	Event event;
	try {
	event = startQuery(query).get(0);
	} catch (IndexOutOfBoundsException e) {
	logger.error(e.getMessage());
	return null;
	}
	return event;
	}

	private List<Event> startQuery(String query) {

	logger.debug(String.format("Query: %s", query.replace("\n", " ")));

	AthenaClient athenaClient = athenaClientFactory.createClient();
	String queryExecutionId = athenaCommon.submitAthenaQuery(athenaClient, query);
	athenaCommon.waitForQueryToComplete(athenaClient, queryExecutionId);
	List<Event> events = processResultRows(athenaClient, queryExecutionId);
	athenaClient.close();

	return events;
	}

	private List<Event> processResultRows(AthenaClient athenaClient, String queryExecutionId) {

	List<Event> events = new ArrayList<>();

	try {
	// Max Results can be set but if it's not set,
	// it will choose the maximum page size
	GetQueryResultsRequest getQueryResultsRequest = GetQueryResultsRequest.builder()
	.queryExecutionId(queryExecutionId).build();

	GetQueryResultsIterable getQueryResultsResults = athenaClient.getQueryResultsPaginator(getQueryResultsRequest);

	List<Row> rows;

	DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.S");

	for (GetQueryResultsResponse result : getQueryResultsResults) {
	rows = result.resultSet().rows();

	for (Row myRow : rows.subList(1, rows.size())) { // skip first row – column names
	List<Datum> allData = myRow.data();
	Event event = new Event();
	event.setId(parseInt(allData.get(0).varCharValue()));
	event.setVenueId(parseInt(allData.get(1).varCharValue()));
	event.setCatId(parseInt(allData.get(2).varCharValue()));
	event.setDateId(parseInt(allData.get(3).varCharValue()));
	event.setName(allData.get(4).varCharValue());
	event.setStartTime(LocalDateTime.parse(allData.get(5).varCharValue(), formatter));
	events.add(event);
	}
	}
	} catch (AthenaException e) {
	logger.error(e.getMessage());
	}
	return events;
	}
	}

view raw EventServiceImp.java hosted with ❤ by GitHub

Using Spring’s common constructor-based Dependency Injection (DI) method (aka constructor injection), the service auto-wires an instance of the AthenaClientFactory interface. Note that we are auto-wiring the service interface, not the service implementation, allowing us to wire in a different implementation if desired, such as for testing.

The service calls the AthenaClientFactoryclass’s createClient() method, which returns a connection to Amazon Athena using one of several available authentication methods. The authentication scheme will depend on where the service is deployed and how you want to securely connect to AWS. Some options include environment variables, local AWS profile, EC2 instance profile, or token from the web identity provider.

	return AthenaClient.builder()
	.credentialsProvider(EnvironmentVariableCredentialsProvider.create())
	.build();

view raw AthenaClient.builder.java hosted with ❤ by GitHub

The service class transforms the payload returned by an instance of GetQueryResultsResponse into an ordered collection (also known as a sequence), List<E>, where E represents a POJO. For example, with the data lake’srefined_tickit_public_event table, the service returns a List<Event>. This pattern repeats itself for tables, views, prepared statements, and named queries. Column data types can be transformed and formatted on the fly, new columns added, and existing columns skipped.

	List<Row> rows;

	DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.S");

	for (GetQueryResultsResponse result : getQueryResultsResults) {
	rows = result.resultSet().rows();

	for (Row myRow : rows.subList(1, rows.size())) { // skip first row – column names
	List<Datum> allData = myRow.data();
	Event event = new Event();
	event.setId(parseInt(allData.get(0).varCharValue()));
	event.setVenueId(parseInt(allData.get(1).varCharValue()));
	event.setCatId(parseInt(allData.get(2).varCharValue()));
	event.setDateId(parseInt(allData.get(3).varCharValue()));
	event.setName(allData.get(4).varCharValue());
	event.setStartTime(LocalDateTime.parse(allData.get(5).varCharValue(), formatter));
	events.add(event);
	}
	}

view raw EventServiceImpSnippet.java hosted with ❤ by GitHub

For each endpoint defined in the Controller class, for example, get(), findAll(), and FindById(), there is a corresponding method in the Service class. Below, we see an example of the findAll() method in the SalesByCategoryServiceImp service class. This method corresponds to the identically named method in the SalesByCategoryController controller class. Each of these service methods follows a similar pattern of constructing a dynamic Athena SQL query based on input parameters, which is passed to Athena through the AthenaClient service client using an instance of GetQueryResultsRequest.

	public List<SalesByCategory> findAll(String calendarDate, Integer limit, Integer offset) {

	if (limit == null \|\| limit < 1 \|\| limit > configProperties.getLimit()) {
	limit = configProperties.getLimit();
	}

	if (offset == null \|\| offset < 1) {
	offset = 0;
	}

	String whereClause = "WHERE caldate IS NOT NULL";
	if (calendarDate != null) {
	whereClause = whereClause + " AND caldate=date('" + calendarDate + "')";
	}

	String query = String.format("""
	SELECT *
	FROM tickit_sales_by_day_and_category
	%s
	OFFSET %s
	LIMIT %s;""", whereClause, offset, limit);

	return startQuery(query);
	}

view raw SalesByCategoryServiceImpSnippet.java hosted with ❤ by GitHub

Controller Class

Lastly, there is a corresponding Controller class for each API endpoint in the Spring Boot RESTful Web Service. The controller class uses Spring Framework’s @RestController annotation. According to the documentation, this annotation is a convenience annotation that is itself annotated with @Controller and @ResponseBody. Types that carry this annotation are treated as controllers where @RequestMapping methods assume @ResponseBody semantics by default.

The controller class takes a dependency on the corresponding service class application component using constructor-based Dependency Injection (DI). Like the service example above, we are auto-wiring the service interface, not the service implementation.

	package com.example.athena.tickit.controller;

	import com.example.athena.tickit.model.saas.Event;
	import com.example.athena.tickit.service.EventService;
	import org.springframework.beans.factory.annotation.Autowired;
	import org.springframework.http.HttpStatus;
	import org.springframework.http.ResponseEntity;
	import org.springframework.web.bind.annotation.*;

	import java.util.List;

	@RestController
	@RequestMapping(value = "/events")
	public class EventController {

	private final EventService service;

	@Autowired
	public EventController(EventService service) {
	this.service = service;
	}

	@RequestMapping(method = RequestMethod.GET)
	public ResponseEntity<List<Event>> findAll(
	@RequestParam(required = false) Integer limit,
	@RequestParam(required = false) Integer offset
	) {

	List<Event> events = service.findAll(limit, offset);
	if (events.size() == 0) {
	return ResponseEntity.status(HttpStatus.NOT_FOUND).body(null);
	}
	return ResponseEntity.status(HttpStatus.OK).body(events);
	}

	@RequestMapping(value = "/{id}", method = RequestMethod.GET)
	public ResponseEntity<Event> findById(@PathVariable("id") int id) {

	Event event = service.findById(id);
	if (event == null) {
	return ResponseEntity.status(HttpStatus.NOT_FOUND).body(null);
	}
	return ResponseEntity.status(HttpStatus.OK).body(event);
	}

	}

view raw EventController.java hosted with ❤ by GitHub

The controller is responsible for serializing the ordered collection of POJOs into JSON and returning that JSON payload in the body of the HTTP response to the initial HTTP request.

Querying Views

In addition to querying AWS Glue Data Catalog tables (aka Athena tables), we also query views. According to the documentation, a view in Amazon Athena is a logical table, not a physical table. Therefore, the query that defines a view runs each time the view is referenced in a query.

For convenience, each time the Spring Boot service starts, the main AthenaApplication class calls the View.java class’s CreateView() method to check for the existence of the view, view_tickit_sales_by_day_and_category. If the view does not exist, it is created and becomes accessible to all application end-users. The view is queried through the service’s /salesbycategory endpoint.

Java class relationships related to querying the Amazon Athena view

This confirm-or-create pattern is repeated for the prepared statement in the main AthenaApplication class (detailed in the next section).

	package com.example.athena;

	import com.example.athena.common.PreparedStatement;
	import com.example.athena.common.View;
	import com.example.athena.config.ConfigProperties;
	import org.springframework.beans.factory.annotation.Autowired;
	import org.springframework.boot.SpringApplication;
	import org.springframework.boot.autoconfigure.SpringBootApplication;
	import org.springframework.boot.context.properties.EnableConfigurationProperties;
	import org.springframework.context.annotation.Bean;
	import org.springframework.web.servlet.config.annotation.CorsRegistry;
	import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;

	@SpringBootApplication
	@EnableConfigurationProperties(ConfigProperties.class)
	public class AthenaApplication {
	private final PreparedStatement preparedStatement;

	private final View view;

	@Autowired
	public AthenaApplication(PreparedStatement preparedStatement, View view) {
	this.preparedStatement = preparedStatement;
	this.view = view;
	}

	public static void main(String[] args) {
	SpringApplication.run(AthenaApplication.class, args);
	}

	@Bean
	void CreatePreparedStatement() {
	preparedStatement.CreatePreparedStatement();
	}

	@Bean
	void createView() {
	view.CreateView();
	}

	@Bean
	public WebMvcConfigurer corsConfigurer() {
	return new WebMvcConfigurer() {
	@Override
	public void addCorsMappings(CorsRegistry registry) {
	registry.addMapping("/*").allowedOrigins("");
	}
	};
	}

	}

view raw AthenaApplication.java hosted with ❤ by GitHub

Below, we see the View class called by the service at startup.

	package com.example.athena.common;

	import com.example.athena.config.ConfigProperties;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;
	import org.springframework.beans.factory.annotation.Autowired;
	import org.springframework.stereotype.Component;
	import software.amazon.awssdk.services.athena.AthenaClient;
	import software.amazon.awssdk.services.athena.model.GetTableMetadataRequest;
	import software.amazon.awssdk.services.athena.model.GetTableMetadataResponse;
	import software.amazon.awssdk.services.athena.model.MetadataException;

	@Component
	public class View {

	private static final Logger logger = LoggerFactory.getLogger(View.class);

	private final AthenaClientFactory athenaClientFactoryImp;

	private final ConfigProperties configProperties;

	private final AthenaCommon athenaCommon;

	@Autowired
	public View(AthenaClientFactory athenaClientFactoryImp,
	ConfigProperties configProperties,
	AthenaCommon athenaCommon) {
	this.athenaClientFactoryImp = athenaClientFactoryImp;
	this.configProperties = configProperties;
	this.athenaCommon = athenaCommon;
	}

	public void CreateView() {
	String viewName = "view_tickit_sales_by_day_and_category";

	String createViewSqlStatement = String.format("""
	CREATE VIEW %s AS
	SELECT cast(d.caldate AS DATE) AS caldate,
	c.catgroup,
	c.catname,
	sum(round(cast(s.pricepaid AS DECIMAL(8,2)) * s.qtysold, 2)) AS saleamount,
	sum(cast(s.commission AS DECIMAL(8,2))) AS commission
	FROM refined_tickit_public_sales AS s
	LEFT JOIN refined_tickit_public_event AS e ON e.eventid = s.eventid
	LEFT JOIN refined_tickit_public_date AS d ON d.dateid = s.dateid
	LEFT JOIN refined_tickit_public_category AS c ON c.catid = e.catid
	GROUP BY caldate,
	catgroup,
	catname
	ORDER BY caldate,
	catgroup,
	catname;""", viewName);

	try (AthenaClient athenaClient = athenaClientFactoryImp.createClient()) {
	try {
	GetTableMetadataResponse getPreparedStatementRequest = getGetTableMetadataResponse(viewName, athenaClient);
	logger.debug(String.format("View already exists: %s", getPreparedStatementRequest.tableMetadata().name()));
	} catch (MetadataException e) { // View does not exist
	String queryExecutionId = athenaCommon.submitAthenaQuery(athenaClient, createViewSqlStatement);
	athenaCommon.waitForQueryToComplete(athenaClient, queryExecutionId);

	// Confirm View was created
	GetTableMetadataResponse getPreparedStatementRequest = getGetTableMetadataResponse(viewName, athenaClient);
	logger.debug(String.format("View created successfully: %s", getPreparedStatementRequest.tableMetadata().name()));
	}
	}
	}

	private GetTableMetadataResponse getGetTableMetadataResponse(String viewName, AthenaClient athenaClient) {
	GetTableMetadataRequest getTableMetadataRequest = GetTableMetadataRequest.builder()
	.catalogName(configProperties.getCatalog())
	.databaseName(configProperties.getDatabase())
	.tableName(viewName)
	.build();

	return athenaClient.getTableMetadata(getTableMetadataRequest);
	}
	}

view raw View.java hosted with ❤ by GitHub

Aside from the fact the /salesbycategory endpoint queries a view, everything else is identical to querying a table. This endpoint uses the same model-service-controller pattern.

Executing Prepared Statements

According to the documentation, you can use the Athena parameterized query feature to prepare statements for repeated execution of the same query with different query parameters. The prepared statement used by the service, tickit_sales_by_seller, accepts a single parameter, the ID of the seller (sellerid). The prepared statement is executed using the /salesbyseller endpoint. This scenario simulates an end-user of the analytics application who wants to retrieve enriched sales information about their sales.

	package com.example.athena.common;

	import com.example.athena.config.ConfigProperties;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;
	import org.springframework.beans.factory.annotation.Autowired;
	import org.springframework.stereotype.Component;
	import software.amazon.awssdk.services.athena.AthenaClient;
	import software.amazon.awssdk.services.athena.model.CreatePreparedStatementRequest;
	import software.amazon.awssdk.services.athena.model.GetPreparedStatementRequest;
	import software.amazon.awssdk.services.athena.model.GetPreparedStatementResponse;
	import software.amazon.awssdk.services.athena.model.ResourceNotFoundException;

	@Component
	public class PreparedStatement {

	private static final Logger logger = LoggerFactory.getLogger(PreparedStatement.class);

	private final AthenaClientFactory athenaClientFactoryImp;

	private final ConfigProperties configProperties;

	@Autowired
	public PreparedStatement(AthenaClientFactory athenaClientFactoryImp, ConfigProperties configProperties) {
	this.athenaClientFactoryImp = athenaClientFactoryImp;
	this.configProperties = configProperties;
	}

	public void CreatePreparedStatement() {
	String preparedStatementName = "tickit_sales_by_seller";

	String preparedStatementSql = """
	SELECT cast(d.caldate AS DATE) AS caldate,
	s.pricepaid,
	s.qtysold,
	round(cast(s.pricepaid AS DECIMAL(8,2)) * s.qtysold, 2) AS saleamount,
	cast(s.commission AS DECIMAL(8,2)) AS commission,
	round((cast(s.commission AS DECIMAL(8,2)) / (cast(s.pricepaid AS DECIMAL(8,2)) * s.qtysold)) * 100, 2) AS commissionprcnt,
	e.eventname,
	concat(u1.firstname, ' ', u1.lastname) AS seller,
	concat(u2.firstname, ' ', u2.lastname) AS buyer,
	c.catgroup,
	c.catname
	FROM refined_tickit_public_sales AS s
	LEFT JOIN refined_tickit_public_listing AS l ON l.listid = s.listid
	LEFT JOIN refined_tickit_public_users AS u1 ON u1.userid = s.sellerid
	LEFT JOIN refined_tickit_public_users AS u2 ON u2.userid = s.buyerid
	LEFT JOIN refined_tickit_public_event AS e ON e.eventid = s.eventid
	LEFT JOIN refined_tickit_public_date AS d ON d.dateid = s.dateid
	LEFT JOIN refined_tickit_public_category AS c ON c.catid = e.catid
	WHERE s.sellerid = ?
	ORDER BY caldate,
	eventname;""";

	try (AthenaClient athenaClient = athenaClientFactoryImp.createClient()) {
	try {
	GetPreparedStatementResponse getPreparedStatementResponse = getGetPreparedStatementResponse(preparedStatementName, athenaClient);
	logger.debug(String.format("Prepared statement already exists: %s", getPreparedStatementResponse.preparedStatement().statementName()));
	} catch (ResourceNotFoundException e) { // PreparedStatement does not exist
	CreatePreparedStatementRequest createPreparedStatementRequest = CreatePreparedStatementRequest.builder()
	.statementName(preparedStatementName)
	.description("Returns all sales by seller based on the seller's userid")
	.workGroup(configProperties.getWorkGroup())
	.queryStatement(preparedStatementSql).build();
	athenaClient.createPreparedStatement(createPreparedStatementRequest);

	// Confirm PreparedStatement was created
	GetPreparedStatementResponse getPreparedStatementResponse = getGetPreparedStatementResponse(preparedStatementName, athenaClient);
	logger.debug(String.format("Prepared statement created successfully: %s", getPreparedStatementResponse.preparedStatement().statementName()));
	}
	}
	}

	private GetPreparedStatementResponse getGetPreparedStatementResponse(String preparedStatementName, AthenaClient athenaClient) {
	GetPreparedStatementRequest getPreparedStatementRequest = GetPreparedStatementRequest.builder()
	.statementName(preparedStatementName)
	.workGroup(configProperties.getWorkGroup()).build();

	return athenaClient.getPreparedStatement(getPreparedStatementRequest);
	}
	}

view raw PreparedStatement.java hosted with ❤ by GitHub

The pattern of querying data is similar to tables and views, except instead of using the common SELECT...FROM...WHERE SQL query pattern, we use the EXECUTE...USING pattern.

	package com.example.athena.tickit.service;

	import com.example.athena.common.AthenaClientFactory;
	import com.example.athena.common.AthenaCommon;
	import com.example.athena.tickit.model.resultsets.SaleBySeller;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;
	import org.springframework.beans.factory.annotation.Autowired;
	import org.springframework.stereotype.Service;
	import software.amazon.awssdk.services.athena.AthenaClient;
	import software.amazon.awssdk.services.athena.model.*;
	import software.amazon.awssdk.services.athena.paginators.GetQueryResultsIterable;

	import java.math.BigDecimal;
	import java.time.LocalDate;
	import java.util.ArrayList;
	import java.util.List;

	@Service
	public class SaleBySellerServiceImp implements SaleBySellerService {

	private static final Logger logger = LoggerFactory.getLogger(SaleBySellerServiceImp.class);

	private final AthenaClientFactory athenaClientFactory;

	private final AthenaCommon athenaCommon;

	@Autowired
	public SaleBySellerServiceImp(AthenaClientFactory athenaClientFactory, AthenaCommon athenaCommon) {
	this.athenaClientFactory = athenaClientFactory;
	this.athenaCommon = athenaCommon;
	}

	public List<SaleBySeller> find(int id) {

	String query = String.format("""
	EXECUTE tickit_sales_by_seller USING %s;""", id);

	return startQuery(query);
	}

	private List<SaleBySeller> startQuery(String query) {

	logger.debug(String.format("Query: %s", query.replace("\n", " ")));

	AthenaClient athenaClient = athenaClientFactory.createClient();
	String queryExecutionId = athenaCommon.submitAthenaQuery(athenaClient, query);
	athenaCommon.waitForQueryToComplete(athenaClient, queryExecutionId);
	List<SaleBySeller> saleBySellers = processResultRows(athenaClient, queryExecutionId);
	athenaClient.close();

	return saleBySellers;
	}

	private List<SaleBySeller> processResultRows(AthenaClient athenaClient, String queryExecutionId) {

	List<SaleBySeller> saleBySellers = new ArrayList<>();

	try {
	GetQueryResultsRequest getQueryResultsRequest = GetQueryResultsRequest.builder()
	.queryExecutionId(queryExecutionId).build();

	GetQueryResultsIterable getQueryResultsResults = athenaClient.getQueryResultsPaginator(getQueryResultsRequest);

	List<Row> rows;

	for (GetQueryResultsResponse result : getQueryResultsResults) {
	rows = result.resultSet().rows();

	for (Row myRow : rows.subList(1, rows.size())) { // skip first row – column names
	List<Datum> allData = myRow.data();
	SaleBySeller saleBySeller = new SaleBySeller();
	saleBySeller.setCalDate(LocalDate.parse(allData.get(0).varCharValue()));
	saleBySeller.setPricePaid(new BigDecimal(allData.get(1).varCharValue()));
	saleBySeller.setQtySold(Integer.parseInt(allData.get(2).varCharValue()));
	saleBySeller.setSaleAmount(new BigDecimal(allData.get(3).varCharValue()));
	saleBySeller.setCommission(new BigDecimal(allData.get(4).varCharValue()));
	saleBySeller.setCommissionPrcnt(Double.valueOf(allData.get(5).varCharValue()));
	saleBySeller.setEventName(allData.get(6).varCharValue());
	saleBySeller.setSeller(allData.get(7).varCharValue());
	saleBySeller.setBuyer(allData.get(8).varCharValue());
	saleBySeller.setCatGroup(allData.get(9).varCharValue());
	saleBySeller.setCatName(allData.get(10).varCharValue());
	saleBySellers.add(saleBySeller);
	}
	}
	} catch (AthenaException e) {
	logger.error(e.getMessage());
	}
	return saleBySellers;
	}
	}

view raw SaleBySellerServiceImp.java hosted with ❤ by GitHub

For example, to execute the prepared statement for a seller with an ID of 3, we would use EXECUTE tickit_sales_by_seller USING 3;. We pass the seller’s ID of 3 as a path parameter similar to other endpoints exposed by the service: /v1/salesbyseller/3.

Sales by seller query results from Athena using the seller’s ID as a parameter for the prepared statement

Again, aside from the fact the /salesbyseller endpoint executes a prepared statement and passes a parameter; everything else is identical to querying a table or a view, using the same model-service-controller pattern.

Working with Named Queries

In addition to tables, views, and prepared statements, Athena has the concept of saved queries, referred to as named queries in the Athena API and when using AWS CloudFormation. You can use the Athena console or API to save, edit, run, rename, and delete queries. The queries are persisted using a NamedQueryId, a unique identifier (UUID) of the query. You must reference the NamedQueryId when working with existing named queries.

Example of saved query (named query) used in this post

There are multiple ways to use and reuse existing named queries programmatically. For this demonstration, I created the named query, buyer_likes_by_category, in advance and then stored the resulting NamedQueryId as an application property, injected at runtime or kubernetes deployment time through a local environment variable.

	package com.example.athena.tickit.service;

	import com.example.athena.common.AthenaClientFactory;
	import com.example.athena.common.AthenaCommon;
	import com.example.athena.config.ConfigProperties;
	import com.example.athena.tickit.model.resultsets.BuyerLikesByCategory;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;
	import org.springframework.beans.factory.annotation.Autowired;
	import org.springframework.stereotype.Service;
	import software.amazon.awssdk.services.athena.AthenaClient;
	import software.amazon.awssdk.services.athena.model.*;
	import software.amazon.awssdk.services.athena.paginators.GetQueryResultsIterable;

	import java.util.ArrayList;
	import java.util.List;

	import static java.lang.Integer.parseInt;

	@Service
	public class BuyerLikesByCategoryServiceImp implements BuyersLikesByCategoryService {

	private static final Logger logger = LoggerFactory.getLogger(BuyerLikesByCategoryServiceImp.class);

	private final ConfigProperties configProperties;

	private final AthenaClientFactory athenaClientFactory;

	private final AthenaCommon athenaCommon;

	@Autowired
	public BuyerLikesByCategoryServiceImp(ConfigProperties configProperties, AthenaClientFactory athenaClientFactory, AthenaCommon athenaCommon) {
	this.configProperties = configProperties;
	this.athenaClientFactory = athenaClientFactory;
	this.athenaCommon = athenaCommon;
	}

	public List<BuyerLikesByCategory> get() {
	return getNamedQueryResults(configProperties.getNamedQueryId());
	}

	private List<BuyerLikesByCategory> getNamedQueryResults(String queryId) {

	logger.debug(String.format("NamedQueryId: %s", queryId));

	AthenaClient athenaClient = athenaClientFactory.createClient();
	GetNamedQueryRequest getNamedQueryRequest = GetNamedQueryRequest.builder()
	.namedQueryId(queryId)
	.build();
	GetNamedQueryResponse getNamedQueryResponse = athenaClient.getNamedQuery(getNamedQueryRequest);
	String queryExecutionId = athenaCommon.submitAthenaQuery(athenaClient, getNamedQueryResponse.namedQuery().queryString());
	athenaCommon.waitForQueryToComplete(athenaClient, queryExecutionId);
	List<BuyerLikesByCategory> buyerLikesByCategories = processResultRows(athenaClient, queryExecutionId);
	athenaClient.close();

	return buyerLikesByCategories;
	}

	private List<BuyerLikesByCategory> processResultRows(AthenaClient athenaClient, String queryExecutionId) {

	List<BuyerLikesByCategory> buyerLikesByCategories = new ArrayList<>();

	try {
	// Max Results can be set but if it's not set,
	// it will choose the maximum page size
	GetQueryResultsRequest getQueryResultsRequest = GetQueryResultsRequest.builder()
	.queryExecutionId(queryExecutionId).build();

	GetQueryResultsIterable getQueryResultsResults = athenaClient.getQueryResultsPaginator(getQueryResultsRequest);

	List<Row> rows;

	for (GetQueryResultsResponse result : getQueryResultsResults) {
	rows = result.resultSet().rows();

	for (Row myRow : rows.subList(1, rows.size())) { // skip first row – column names
	List<Datum> allData = myRow.data();
	BuyerLikesByCategory buyerLikesByCategory = new BuyerLikesByCategory();
	buyerLikesByCategory.setSports(parseInt(allData.get(0).varCharValue()));
	buyerLikesByCategory.setTheatre(parseInt(allData.get(1).varCharValue()));
	buyerLikesByCategory.setConcerts(parseInt(allData.get(2).varCharValue()));
	buyerLikesByCategory.setJazz(parseInt(allData.get(3).varCharValue()));
	buyerLikesByCategory.setClassical(parseInt(allData.get(4).varCharValue()));
	buyerLikesByCategory.setOpera(parseInt(allData.get(5).varCharValue()));
	buyerLikesByCategory.setRock(parseInt(allData.get(6).varCharValue()));
	buyerLikesByCategory.setVegas(parseInt(allData.get(7).varCharValue()));
	buyerLikesByCategory.setBroadway(parseInt(allData.get(8).varCharValue()));
	buyerLikesByCategory.setMusicals(parseInt(allData.get(9).varCharValue()));
	buyerLikesByCategories.add(buyerLikesByCategory);
	}
	}
	} catch (AthenaException e) {
	logger.error(e.getMessage());
	}
	return buyerLikesByCategories;
	}
	}

view raw BuyerLikesByCategoryServiceImp.java hosted with ❤ by GitHub

Alternately, you might iterate through a list of named queries to find one that matches the name at startup. However, this method would undoubtedly impact service performance, startup time, and cost. Lastly, you could use a method like NamedQuery() included in the unused NamedQuery class at startup, similar to the view and prepared statement. That named query’s unique NamedQueryId would be persisted as a system property, referencable by the service class. The downside is that you would create a duplicate of the named query each time you start the service. Therefore, this method is also not recommended.

Configuration

Two components responsible for persisting configuration for the Spring Boot service are the application.yml properties file and ConfigProperties class. The class uses Spring Framework’s @ConfigurationProperties annotation. According to the documentation, this annotation is used for externalized configuration. Add this to a class definition or a @Bean method in a @Configuration class if you want to bind and validate some external Properties (e.g., from a .properties or .yml file). Binding is performed by calling setters on the annotated class or, if @ConstructorBinding in use, by binding to the constructor parameters.

The @ConfigurationProperties annotation includes the prefix of athena. This value corresponds to the athena prefix in the the application.yml properties file. The fields in the ConfigProperties class are bound to the properties in the the application.yml. For example, the property, namedQueryId, is bound to the property, athena.named.query.id. Further, that property is bound to an external environment variable, NAMED_QUERY_ID. These values could be supplied from an external configuration system, a Kubernetes secret, or external secrets management system.

	spring:
	profiles:
	active: dev
	server:
	port: 8080
	servlet:
	contextPath: /v1
	athena:
	region: us-east-1
	workgroup: primary
	catalog: AwsDataCatalog
	database: tickit_demo
	limit: 25
	client-execution-timeout: 100000
	retry-sleep: 1000
	results-bucket: ${RESULTS_BUCKET}
	named-query-id: ${NAMED_QUERY_ID}
	—
	spring:
	config:
	activate:
	on-profile: dev
	logging:
	level:
	root: DEBUG
	management:
	endpoints:
	web:
	exposure:
	include: '*'
	jmx:
	exposure:
	include: '*'
	—
	spring:
	config:
	activate:
	on-profile: prod
	logging:
	level:
	root: INFO
	management:
	endpoints:
	web:
	exposure:
	include: health, prometheus
	jmx:
	exposure:
	include: health

view raw application.yml hosted with ❤ by GitHub

AWS IAM: Authentication and Authorization

For the Spring Boot service to interact with Amazon Athena, AWS Glue, and Amazon S3, you need to establish an AWS IAM Role, which the service assumes once authenticated. The Role must be associated with an attached IAM Policy containing the requisite Athena, Glue, and S3 permissions. For development, the service uses a policy similar to the one shown below. Please note this policy is broader than recommended for Production; it does not represent the security best practice of least privilege. In particular, the use of the overly-broad * for Resources should be strictly avoided when creating policies.

	{
	"Version": "2012-10-17",
	"Statement": [
	{
	"Effect": "Allow",
	"Action": [
	"athena:StartQueryExecution",
	"athena:CreatePreparedStatement",
	"athena:ListPreparedStatements",
	"glue:CreateTable",
	"athena:CreateNamedQuery",
	"athena:ListNamedQueries",
	"athena:GetTableMetadata",
	"athena:GetPreparedStatement",
	"athena:GetQueryResults",
	"athena:GetQueryExecution",
	"athena:GetNamedQuery"
	],
	"Resource": [
	"*"
	]
	},
	{
	"Effect": "Allow",
	"Action": [
	"glue:BatchGetPartition",
	"glue:GetTable"
	],
	"Resource": [
	"*"
	]
	},
	{
	"Effect": "Allow",
	"Action": [
	"s3:GetBucketLocation",
	"s3:GetObject",
	"s3:ListBucket",
	"s3:ListBucketMultipartUploads",
	"s3:ListMultipartUploadParts",
	"s3:AbortMultipartUpload",
	"s3:CreateBucket",
	"s3:PutObject",
	"s3:PutBucketPublicAccessBlock"
	],
	"Resource": [
	"arn:aws:s3:::aws-athena-query-results-*"
	]
	},
	{
	"Effect": "Allow",
	"Action": [
	"s3:GetObject",
	"s3:ListBucket"
	],
	"Resource": [
	"arn:aws:s3:::date-lake-demo-*"
	]
	},
	{
	"Effect": "Allow",
	"Action": [
	"s3:ListBucket",
	"s3:GetBucketLocation",
	"s3:ListAllMyBuckets"
	],
	"Resource": [
	"*"
	]
	}
	]
	}

view raw AthenaSpringAppPolicy.json hosted with ❤ by GitHub

In addition to the authorization granted by the IAM Policy, AWS Lake Formation can be used with Amazon S3, AWS Glue, and Amazon Athena to grant fine-grained database-, table-, column-, and row-level access to datasets.

Swagger UI and the OpenAPI Specification

The easiest way to view and experiment with all the endpoints available through the controller classes is using the Swagger UI, included in the example Spring Boot service, by way of the springdoc-openapi Java library. The Swagger UI is accessed at /v1/swagger-ui/index.html.

Swagger UI showing endpoints exposed by the service’s controller classes

The OpenAPI Specification (formerly Swagger Specification) is an API description format for REST APIs. The /v1/v3/api-docs endpoint allows you to generate an OpenAPI v3 specification file. The OpenAPI file describes the entire API.

Spring Boot service’s OpenAPI v3 specification

The OpenAPI v3 specification can be saved as a file and imported into applications like Postman, the API platform for building and using APIs.

Calling the service’s `/users` API endpoint using Postman

Integration Tests

Included in the Spring Boot service’s source code is a limited number of example integration tests, not to be confused with unit tests. Each test class uses Spring Framework’s @SpringBootTest annotation. According to the documentation, this annotation can be specified on a test class that runs Spring Boot-based tests. It provides several features over and above the regular Spring TestContext Framework.

	package com.example.athena;

	import io.restassured.http.ContentType;
	import org.hamcrest.Matchers;
	import org.junit.jupiter.api.Test;
	import org.springframework.boot.test.context.SpringBootTest;

	import static io.restassured.RestAssured.get;
	import static io.restassured.RestAssured.given;
	import static io.restassured.http.ContentType.JSON;

	@SpringBootTest
	class CategoriesResourceTests {

	private static final String ResourcePath = "/v1/categories";

	private static final int resultsetLimit = 25;

	@Test
	void findAll() {
	given()
	.contentType(ContentType.JSON)
	.accept(ContentType.JSON)
	.when()
	.get(ResourcePath)
	.then()
	.assertThat()
	.contentType(JSON)
	.statusCode(200)
	.body("$.size()", Matchers.greaterThanOrEqualTo(1))
	.body("$.size()", Matchers.lessThanOrEqualTo(resultsetLimit));
	}

	@Test
	void findAllWithLimit() {
	given()
	.contentType(ContentType.JSON)
	.accept(ContentType.JSON)
	.queryParam("limit", 3)
	.when()
	.get(ResourcePath)
	.then()
	.assertThat()
	.contentType(JSON)
	.statusCode(200)
	.body("$.size()", Matchers.equalTo(3));
	}

	@Test
	void findAllWithOffset() {
	given()
	.contentType(ContentType.JSON)
	.accept(ContentType.JSON)
	.queryParam("offset", 2)
	.when()
	.get(ResourcePath)
	.then()
	.assertThat()
	.contentType(JSON)
	.statusCode(200)
	.body("$.size()", Matchers.greaterThanOrEqualTo(1))
	.body("$.size()", Matchers.lessThanOrEqualTo(resultsetLimit));
	}

	@Test
	void findAllWithLimitAndOffset() {
	given()
	.contentType(ContentType.JSON)
	.accept(ContentType.JSON)
	.queryParam("limit", 3)
	.queryParam("offset", 2)
	.when()
	.get(ResourcePath)
	.then()
	.assertThat()
	.contentType(JSON)
	.statusCode(200)
	.body("$.size()", Matchers.equalTo(3));
	}

	@Test
	void findById() {
	// Get the first 'id' available
	int id = get(ResourcePath + "?limit=1")
	.then()
	.extract()
	.path("[0].id");

	given()
	.contentType(ContentType.JSON)
	.accept(ContentType.JSON)
	.when()
	.get(ResourcePath + "/{id}", id)
	.then()
	.assertThat()
	.contentType(JSON)
	.statusCode(200)
	.body("id", Matchers.equalTo(id));
	}
	}

view raw CategoriesResourceTests.java hosted with ❤ by GitHub

The integration tests use Rest Assured’s given-when-then pattern of testing, made popular as part of Behavior-Driven Development (BDD). In addition, each test uses the JUnit’s @Test annotation. According to the documentation, this annotation signals that the annotated method is a test method. Therefore, methods using this annotation must not be private or static and must not return a value.

	@Test
	void findById() {
	// Get the first 'id' available
	int id = get(ResourcePath + "?limit=1")
	.then()
	.extract()
	.path("[0].id");

	given()
	.contentType(ContentType.JSON)
	.accept(ContentType.JSON)
	.when()
	.get(ResourcePath + "/{id}", id)
	.then()
	.assertThat()
	.contentType(JSON)
	.statusCode(200)
	.body("id", Matchers.equalTo(id));
	}

view raw CategoriesResourceTestsSnippet.java hosted with ❤ by GitHub

Run the integration tests using Gradle from the project’s root: ./gradlew clean build test. A detailed ‘Test Summary’ is produced in the project’s build directory as HTML for easy review.

Load Testing the Service

In Production, the Spring Boot service will need to handle multiple concurrent users executing queries against Amazon Athena.

Athena’s Recent Queries console shows multi concurrent queries being queued and executed

We could use various load testing tools to evaluate the service’s ability to handle multiple concurrent users. One of the simplest is my favorite go-based utility, hey, which sends load to a URL using a provided number of requests in the provided concurrency level and prints stats. It also supports HTTP2 endpoints. So, for example, we could execute 500 HTTP requests with a concurrency level of 25 against the Spring Boot service’s /users endpoint using hey. The post’s integration tests were run against three Kubernetes replica pods of the service deployed to Amazon EKS.

hey -n 500 -c 25 -T "application/json;charset=UTF-8" \
  -h2 https://athena.example-api.com/v1/users

From Athena’s Recent Queries console, we see many simultaneous queries being queued and executed by a hey through the Spring Boot service’s endpoint.

Athena’s Recent Queries console shows simultaneous queries being queued and executed

Metrics

The Spring Boot service implements the micrometer-registry-prometheus extension. The Micrometer metrics library exposes runtime and application metrics. Micrometer defines a core library, providing a registration mechanism for metrics and core metric types. These metrics are exposed by the service’s /v1/actuator/prometheus endpoint.

Using the Micrometer extension, metrics exposed by the /v1/actuator/prometheus endpoint can be scraped and visualized by tools such as Prometheus. Conveniently, AWS offers the fully-managed Amazon Managed Service for Prometheus (AMP), which easily integrates with Amazon EKS.

Graph of HTTP server requests scraped by Prometheus from the Spring Boot service

Using Prometheus as a datasource, we can build dashboards in Grafana to observe the service’s metrics. Like AMP, AWS also offers the fully-managed Amazon Managed Grafana (AMG).

Grafana dashboard showing metrics from Prometheus for Spring Boot service deployed to Amazon EKS

Grafana dashboard showing JVM metrics from Prometheus for Spring Boot service deployed to Amazon EKS

Conclusion

This post taught us how to create a Spring Boot RESTful Web Service, allowing end-user applications to securely query data stored in a data lake on AWS. The service used AWS SDK for Java to access data stored in Amazon S3 through an AWS Glue Data Catalog using Amazon Athena.

This blog represents my own viewpoints and not of my employer, Amazon Web Services (AWS). All product names, logos, and brands are the property of their respective owners. All diagrams and illustrations are property of the author unless otherwise noted.

Amazon Athena, Analytics, AWS, Data Lake, Java, RESTful API, Spring Boot

Leave a comment

The Art of Building Open Data Lakes with Apache Hudi, Kafka, Hive, and Debezium

Posted by Gary A. Stafford in Analytics, AWS, Big Data, Cloud, Python, Software Development, Technology Consulting on December 31, 2021

Build near real-time, open-source data lakes on AWS using a combination of Apache Kafka, Hudi, Spark, Hive, and Debezium

Introduction

In the following post, we will learn how to build a data lake on AWS using a combination of open-source software (OSS), including Red Hat’s Debezium, Apache Kafka, Kafka Connect, Apache Hive, Apache Spark, Apache Hudi, and Hudi DeltaStreamer. We will use fully-managed AWS services to host the datasource, the data lake, and the open-source tools. These services include Amazon RDS, MKS, EKS, EMR, and S3.

This post is an in-depth follow-up to the video demonstration, Building Open Data Lakes on AWS with Debezium and Apache Hudi.

Workflow

As shown in the architectural diagram above, these are the high-level steps in the demonstration’s workflow:

Changes (inserts, updates, and deletes) are made to the datasource, a PostgreSQL database running on Amazon RDS;
Kafka Connect Source Connector, utilizing Debezium and running on Amazon EKS (Kubernetes), continuously reads data from PostgreSQL WAL using Debezium;
Source Connector creates and stores message schemas in Apicurio Registry, also running on Amazon EKS, in Avro format;
Source Connector transforms and writes data in Apache Avro format to Apache Kafka, running on Amazon MSK;
Kafka Connect Sink Connector, using Confluent S3 Sink Connector, reads messages from Kafka topics using schemas from Apicurio Registry;
Sink Connector writes data to Amazon S3 in Apache Avro format;
Apache Spark, using Hudi DeltaStreamer and running on Amazon EMR, reads message schemas from Apicurio Registry;
DeltaStreamer reads raw Avro-format data from Amazon S3;
DeltaStreamer writes data to Amazon S3 as both Copy on Write (CoW) and Merge on Read (MoR) table types;
DeltaStreamer syncs Hudi tables and partitions to Apache Hive running on Amazon EMR;
Queries are executed against Apache Hive Metastore or directly against Hudi tables using Apache Spark, with data returned from Hudi tables in Amazon S3;

The workflow described above actually contains two independent processes running simultaneously. Steps 2–6 represent the first process, the change data capture (CDC) process. Kafka Connect is used to continuously move changes from the database to Amazon S3. Steps 7–10 represent the second process, the data lake ingestion process. Hudi’s DeltaStreamer reads raw CDC data from Amazon S3 and writes the data back to another location in S3 (the data lake) in Apache Hudi table format. When combined, these processes can give us near real-time, incremental data ingestion of changes from the datasource to the Hudi-managed data lake.

Alternatives

This demonstration’s workflow is only one of many possible workflows to achieve similar outcomes. Alternatives include:

Replace self-managed Kafka Connect with the fully-managed Amazon MSK Connect service.
Exchange Amazon EMR for AWS Glue Jobs or AWS Glue Studio and the custom AWS Glue Connector for Apache Hudi to ingest data into Hudi tables.
Replace Apache Hive with AWS Glue Data Catalog, a fully-managed Hive-compatible metastore.
Replace Apicurio Registry with Confluent Schema Registry or AWS Glue Schema Registry.
Exchange the Confluent S3 Sink Connector for the Kafka Connect Sink for Hudi, which could greatly simplify the workflow.
Substitute HoodieMultiTableDeltaStreamer for the HoodieDeltaStreamer utility to quickly ingest multiple tables into Hudi.
Replace Hudi’s AvroDFSSource for the AvroKafkaSource to read directly from Kafka versus Amazon S3, or Hudi’s JdbcSource to read directly from the PostgreSQL database. Hudi has several datasource readers available. Be cognizant of authentication/authorization compatibility/limitations.
Choose either or both Hudi’s Copy on Write (CoW) and Merge on Read (MoR) table types depending on your workload requirements.

Source Code

All source code for this post and the previous posts in this series are open-sourced and located on GitHub. The specific resources used in this post are found in the debezium_hudi_demo directory of the GitHub repository. There are also two copies of the Museum of Modern Art (MoMA) Collection dataset from Kaggle, specifically prepared for this post, located in the moma_data directory. One copy is a nearly full dataset, and the other is a smaller, cost-effective dev/test version.

Kafka Connect

In this demonstration, Kafka Connect runs on Kubernetes, hosted on the fully-managed Amazon Elastic Kubernetes Service (Amazon EKS). Kafka Connect runs the Source and Sink Connectors.

Source Connector

The Kafka Connect Source Connector, source_connector_moma_postgres_kafka.json, used in steps 2–4 of the workflow, utilizes Debezium to continuously read changes to an Amazon RDS for PostgreSQL database. The PostgreSQL database hosts the MoMA Collection in two tables: artists and artworks.

	{
	"connector.class": "io.debezium.connector.postgresql.PostgresConnector",
	"database.hostname": "<your_database_hostname>",
	"database.port": "5432",
	"database.user": "<your_username>",
	"database.password": "<your_password>",
	"database.dbname": "moma",
	"database.server.name": "moma",
	"table.include.list": "public.artists,public.artworks",
	"plugin.name": "pgoutput",
	"key.converter": "io.apicurio.registry.utils.converter.AvroConverter",
	"key.converter.apicurio.registry.url": "http://localhost:8080/apis/registry/v2",
	"key.converter.apicurio.registry.auto-register": "true",
	"key.converter.apicurio.registry.find-latest": "true",
	"value.converter": "io.apicurio.registry.utils.converter.AvroConverter",
	"value.converter.apicurio.registry.url": "http://localhost:8080/apis/registry/v2",
	"value.converter.apicurio.registry.auto-register": "true",
	"value.converter.apicurio.registry.find-latest": "true",
	"transforms": "unwrap",
	"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
	"transforms.unwrap.drop.tombstones": "false",
	"transforms.unwrap.delete.handling.mode": "rewrite",
	"transforms.unwrap.add.fields": "op,db,table,schema,lsn,source.ts_ms"
	}

view raw source_connector_moma_postgres_kafka.json hosted with ❤ by GitHub

The Debezium Connector for PostgreSQL reads record-level insert, update, and delete entries from PostgreSQL’s write-ahead log (WAL). According to the PostgreSQL documentation, changes to data files must be written only after log records describing the changes have been flushed to permanent storage, thus the name, write-ahead log. The Source Connector then creates and stores Apache Avro message schemas in Apicurio Registry also running on Amazon EKS.

Apicurio Registry UI showing Avro-format Kafka message schemas

Apicurio Registry UI showing part of Avro-format Kafka message value schema for artists

Finally, the Source Connector transforms and writes Avro format messages to Apache Kafka running on the fully-managed Amazon Managed Streaming for Apache Kafka (Amazon MSK). Assuming Kafka’s topic.creation.enable property is set to true, Kafka Connect will create any necessary Kafka topics, one per database table.

Below, we see an example of a Kafka message representing an insert of a record with the artist_id 1 in the MoMA Collection database’s artists table. The record was read from the PostgreSQL WAL, transformed, and written to a corresponding Kafka topic, using the Debezium Connector for PostgreSQL. The first version represents the raw data before being transformed by Debezium. Note that the type of operation (_op) indicates a read (r). Possible values include c for create (or insert), u for update, d for delete, and r for read (applies to snapshots).

	{
	"payload": {
	"before": null,
	"after": {
	"artist_id": 1,
	"name": "Robert Arneson",
	"nationality": "American",
	"gender": "Male",
	"birth_year": 1930,
	"death_year": 1992
	},
	"source": {
	"version": "1.7.0.Final",
	"connector": "postgresql",
	"name": "moma",
	"ts_ms": 1640703877051,
	"snapshot": "true",
	"db": "moma",
	"sequence": "[null,\"3668170506336\"]",
	"schema": "public",
	"table": "artists",
	"txId": 217094,
	"lsn": 3668170506336,
	"xmin": null
	},
	"op": "r",
	"ts_ms": 1640703877051,
	"transaction": null
	}
	}

view raw artist_kafka_message_untransformed.json hosted with ❤ by GitHub

The next version represents the same record after being transformed by Debezium using the event flattening single message transformation (unwrap SMT). The final message structure represents the schema stored in Apicurio Registry. The message structure is identical to the structure of the data written to Amazon S3 by the Sink Connector.

	{
	"payload": {
	"artist_id": 1,
	"name": "Robert Arneson",
	"nationality": "American",
	"gender": "Male",
	"birth_year": 1930,
	"death_year": 1992,
	"__op": "r",
	"__db": "moma",
	"__table": "artists",
	"__schema": "public",
	"__lsn": 3668438941792,
	"__source_ts_ms": 1640705109121,
	"__deleted": "false"
	}
	}

view raw artist_kafka_message_transformed.json hosted with ❤ by GitHub

Sink Connector

The Kafka Connect Sink Connector, sink_connector_moma_kafka_s3.json, used in steps 5–6 of the workflow, implements the Confluent S3 Sink Connector. The Sink Connector reads the Avro-format messages from Kafka using the schemas stored in Apicurio Registry. It then writes the data to Amazon S3, also in Apache Avro format, based on the same schemas.

	{
	"connector.class": "io.confluent.connect.s3.S3SinkConnector",
	"tasks.max": 1,
	"topics.regex": "moma.public.(.*)",
	"table.name.format": "${topic}",
	"s3.region": "us-east-1",
	"s3.bucket.name": "<your_data_lake_bucket>",
	"s3.part.size": 5242880,
	"flush.size": 10000,
	"rotate.schedule.interval.ms": 60000,
	"timezone": "UTC",
	"storage.class": "io.confluent.connect.s3.storage.S3Storage",
	"format.class": "io.confluent.connect.s3.format.avro.AvroFormat",
	"schema.generator.class": "io.confluent.connect.storage.hive.schema.DefaultSchemaGenerator",
	"partitioner.class": "io.confluent.connect.storage.partitioner.DefaultPartitioner",
	"schema.compatibility": "NONE",
	"behavior.on.null.values": "ignore",
	"key.converter": "io.apicurio.registry.utils.converter.AvroConverter",
	"key.converter.apicurio.registry.url": "http://localhost:8080/apis/registry/v2",
	"key.converter.apicurio.registry.auto-register": "true",
	"key.converter.apicurio.registry.find-latest": "true",
	"value.converter": "io.apicurio.registry.utils.converter.AvroConverter",
	"value.converter.apicurio.registry.url": "http://localhost:8080/apis/registry/v2",
	"value.converter.apicurio.registry.auto-register": "true",
	"value.converter.apicurio.registry.find-latest": "true"
	}

view raw sink_connector_moma_kafka_s3.json hosted with ❤ by GitHub

Running Kafka Connect

We first start Kafka Connect in the background to be the CDC process.

	bin/connect-distributed.sh \
	config/connect-distributed.properties \
	> /dev/null 2>&1 &

	tail -f logs/connect.log

view raw start_kafka_connect.sh hosted with ❤ by GitHub

Then, deploy the Kafka Connect Source and Sink Connectors using Kafka Connect’s RESTful API. Using the API, we can also confirm the status of the Connectors.

	curl -s -d @"config/source_connector_moma_postgres_kafka.json" \
	-H "Content-Type: application/json" \
	-X PUT http://localhost:8083/connectors/source_connector_moma_postgres_kafka/config \| jq

	curl -s -d @"config/sink_connector_moma_kafka_s3.json" \
	-H "Content-Type: application/json" \
	-X PUT http://localhost:8083/connectors/sink_connector_moma_kafka_s3/config \| jq

	curl -s -X GET http://localhost:8083/connectors \| jq

	curl -s -H "Content-Type: application/json" \
	-X GET http://localhost:8083/connectors/source_connector_moma_postgres_kafka/status \| jq

	curl -s -H "Content-Type: application/json" \
	-X GET http://localhost:8083/connectors/sink_connector_moma_kafka_s3/status \| jq

view raw kafka_connect_connectors.sh hosted with ❤ by GitHub

To confirm the two Kafka topics, moma.public.artists and moma.public.artworks, were created and contain Avro messages, we can use Kafka’s command-line tools.

	# list kafka topics
	bin/kafka-topics.sh –list \
	–bootstrap-server $BBROKERS \
	–command-config config/client-iam.properties

	# read first 5 avro-format (binary) messages from topic
	bin/kafka-console-consumer.sh \
	–topic moma.public.artists \
	–from-beginning \
	–max-messages 5 \
	–property print.value=true \
	–property print.offset=true \
	–bootstrap-server $BBROKERS \
	–consumer.config config/client-iam.properties

view raw confirm_topics_messages.sh hosted with ❤ by GitHub

In the short video-only clip below, we see the process of deploying the Kafka Connect Source and Sink Connectors and confirming they are working as expected.

Deploying and starting the Kafka Connect Source and Sink Connectors

The Sink Connector writes data to Amazon S3 in batches of 10k messages or every 60 seconds (one-minute intervals). These settings are configurable and highly dependent on your requirements, including message volume, message velocity, real-time analytics requirements, and available compute resources.

Amazon S3 objects containing MoMA Collection artwork records from PostgreSQL

Since we will not be querying this raw Avro-format CDC data in Amazon S3 directly, there is no need to catalog this data in Apache Hive or AWS Glue Data Catalog, a fully-managed Hive-compatible metastore.

Apache Hudi

According to the overview, Apache Hudi (pronounced “hoodie”) is the next-generation streaming data lake platform. Apache Hudi brings core warehouse and database functionality to data lakes. Hudi provides tables, transactions, efficient upserts and deletes, advanced indexes, streaming ingestion services, data clustering, compaction optimizations, and concurrency, all while keeping data in open source file formats.

Without Hudi or an equivalent open-source data lake table format such as Apache Iceberg or Databrick’s Delta Lake, most data lakes are just of bunch of unmanaged flat files. Amazon S3 cannot natively maintain the latest view of the data, to the surprise of many who are more familiar with OLTP-style databases or OLAP-style data warehouses.

DeltaStreamer

DeltaStreamer, aka the HoodieDeltaStreamer utility (part of the hudi-utilities-bundle), used in steps 7–10 of the workflow, provides the way to perform streaming ingestion of data from different sources such as Distributed File System (DFS) and Apache Kafka.

Optionally, HoodieMultiTableDeltaStreamer, a wrapper on top of HoodieDeltaStreamer, ingests multiple tables in a single Spark job, into Hudi datasets. Currently, it only supports sequential processing of tables to be ingested and Copy on Write table type.

We are using HoodieDeltaStreamer to write to both Merge on Read (MoR) and Copy on Write (CoW) table types for demonstration purposes only. The MoR table type is a superset of the CoW table type, which stores data using a combination of columnar-based (e.g., Apache Parquet) plus row-based (e.g., Apache Avro) file formats. Updates are logged to delta files and later compacted to produce new versions of columnar files synchronously or asynchronously. Again, the choice of table types depends on your requirements.

Trade-offs between Hudi table types (table courtesy Apache Hudi documentation)

Amazon EMR

For this demonstration, I’ve used the recently released Amazon EMR version 6.5.0 configured with Apache Spark 3.1.2 and Apache Hive 3.1.2. EMR 6.5.0 runs Scala version 2.12.10, Python 3.7.10, and OpenJDK Corretto-8.312. I have included the AWS CloudFormation template and parameters file used to create the EMR cluster, on GitHub.

When choosing Apache Spark, Apache Hive, or Presto on EMR 6.5.0, Apache Hudi release 0.9.0 is automatically installed.

Amazon EMR Master Node showing Apache Hudi related resources

DeltaStreamer Configuration

Below, we see the DeltaStreamer properties file, deltastreamer_artists_apicurio_mor.properties. This properties file is referenced by the Spark job that runs DeltaStreamer, shown next. The file contains properties related to the datasource, the data sink, and Apache Hive. The source of the data for DeltaStreamer is the CDC data written to Amazon S3. In this case, the datasource is the objects located in the /topics/moma.public.artworks/partition=0/ S3 object prefix. The data sink is a Hudi MoR table type in Amazon S3. DeltaStreamer will write Parquet data, partitioned by the artist’s nationality, to the /moma_mor/artists/ S3 object prefix. Lastly, DeltaStreamer will sync all tables and table partitions to Apache Hive, including creating the Hive databases and tables if they do not already exist.

	# Built for demo of Apache Hudi 0.9.0 (EMR 6.5.0) with Apache Hive and SchemaRegistryProvider

	include=base.properties

	hoodie.datasource.hive_sync.assume_date_partitioning=false
	hoodie.datasource.hive_sync.database=moma_mor
	hoodie.datasource.hive_sync.partition_extractor_class=org.apache.hudi.hive.MultiPartKeysValueExtractor
	hoodie.datasource.hive_sync.partition_fields=nationality
	hoodie.datasource.hive_sync.table=artists

	hoodie.datasource.write.drop.partition.columns=true
	hoodie.datasource.write.hive_style_partitioning=true
	hoodie.datasource.write.partitionpath.field=nationality
	hoodie.datasource.write.recordkey.field=artist_id

	hoodie.deltastreamer.schemaprovider.registry.url=http://<your_registry_url:post>/apis/ccompat/v6/subjects/moma.public.artists-value/versions/latest

	hoodie.deltastreamer.source.dfs.root=s3://<your_data_lake_bucket>/topics/moma.public.artists/partition=0/

	# 1,024 * 1,024 * 128 = 134,217,728 (128 MB)
	hoodie.parquet.small.file.limit=134217728

	# https://dacort.dev/posts/updating-partition-values-with-apache-hudi/
	# This is required if we want to ensure we upsert a record, even if the partition changes
	hoodie.index.type=GLOBAL_BLOOM

	# This is required to write the data into the new partition
	# defaults to false in Apache Hudi 0.8.0 (EMR 6.4.0), true in Hudi 0.9.0 (EMR 6.5.0)
	# hoodie.bloom.index.update.partition.path=true

view raw deltastreamer_artists_apicurio.properties hosted with ❤ by GitHub

DeltaStreamer properties file for artists data using MoR table type

Below, we see the equivalent DeltaStreamer properties file for the MoMA artworks, deltastreamer_artworks_apicurio_mor.properties. There are also comparable DeltaStreamer property files for the Hudi CoW tables on GitHub.

	# Built for demo of Apache Hudi 0.9.0 (EMR 6.5.0) with Apache Hive and SchemaRegistryProvider

	include=base.properties

	hoodie.datasource.hive_sync.assume_date_partitioning=false
	hoodie.datasource.hive_sync.database=moma_mor
	hoodie.datasource.hive_sync.partition_extractor_class=org.apache.hudi.hive.MultiPartKeysValueExtractor
	hoodie.datasource.hive_sync.partition_fields=classification
	hoodie.datasource.hive_sync.table=artworks

	hoodie.datasource.write.drop.partition.columns=true
	hoodie.datasource.write.hive_style_partitioning=true
	hoodie.datasource.write.partitionpath.field=classification
	hoodie.datasource.write.recordkey.field=artwork_id

	hoodie.deltastreamer.schemaprovider.registry.url=http://<your_registry_url:post>/apis/ccompat/v6/subjects/moma.public.artworks-value/versions/latest

	hoodie.deltastreamer.source.dfs.root=s3://<your_data_lake_bucket>/topics/moma.public.artworks/partition=0/

	# 1,024 * 1,024 * 128 = 134,217,728 (128 MB)
	hoodie.parquet.small.file.limit=134217728

	# https://dacort.dev/posts/updating-partition-values-with-apache-hudi/
	# This is required if we want to ensure we upsert a record, even if the partition changes
	hoodie.index.type=GLOBAL_BLOOM

	# This is required to write the data into the new partition
	# defaults to false in Apache Hudi 0.8.0 (EMR 6.4.0), true in Hudi 0.9.0 (EMR 6.5.0)
	# hoodie.bloom.index.update.partition.path=true

view raw deltastreamer_artworks_apicurio_mor.properties hosted with ❤ by GitHub

DeltaStreamer properties file for artworks data using MoR table type

All DeltaStreamer property files reference Apicurio Registry for the location of the Avro schemas. The schemas are used by both the Kafka Avro-format messages and the CDC-created Avro-format files in Amazon S3. Due to DeltaStreamer’s coupling with Confluent Schema Registry, as opposed to other registries, we must use Apicurio Registry’s Confluent Schema Registry API (Version 6) compatibility API endpoints (e.g., /apis/ccompat/v6/subjects/moma.public.artists-value/versions/latest) when using the org.apache.hudi.utilities.schema.SchemaRegistryProvider datasource option with DeltaStreamer. According to Apicurio, to provide compatibility with Confluent SerDes (Serializer/Deserializer) and other clients, Apicurio Registry implements the API defined by the Confluent Schema Registry.

Running DeltaStreamer

The properties files are loaded by Spark jobs that call the DeltaStreamer library, using spark-submit. Below, we see an example Spark job that calls the DeltaStreamer class. DeltaStreamer reads the raw Avro-format CDC data from S3 and writes the data using the Hudi MoR table type into the /moma_mor/artists/ S3 object prefix. In this Spark particular job, we are using the continuous option. DeltaStreamer runs in continuous mode using this option, running source-fetch, transform, and write in a loop. We are also using the UPSERT write operation (op). Operation options include UPSERT, INSERT, and BULK_INSERT. This set of options is ideal for inserting ongoing changes to CDC data into Hudi tables. You can run jobs in the foreground or background on EMR’s Master Node or as EMR Steps from the Amazon EMR console.

	export DATA_LAKE_BUCKET="<your_data_lake_bucket_name>"

	# artists data, MoR table type, continuous upserts
	spark-submit \
	–jars /usr/lib/spark/jars/spark-avro.jar,/usr/lib/hudi/hudi-utilities-bundle.jar \
	–conf spark.sql.catalogImplementation=hive \
	–class org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer /usr/lib/hudi/hudi-utilities-bundle.jar \
	–table-type MERGE_ON_READ \
	–source-ordering-field __source_ts_ms \
	–props "s3://${DATA_LAKE_BUCKET}/hudi/deltastreamer_artists_apicurio_mor.properties" \
	–source-class org.apache.hudi.utilities.sources.AvroDFSSource \
	–target-base-path "s3://${DATA_LAKE_BUCKET}/moma/artists_mor/" \
	–target-table moma_mor.artists \
	–schemaprovider-class org.apache.hudi.utilities.schema.SchemaRegistryProvider \
	–enable-sync \
	–continuous \
	–op UPSERT

view raw deltastreamer_artists_continuous.sh hosted with ❤ by GitHub

Below, we see another example DeltaStreamer Spark job that reads the raw Avro-format CDC data from S3 and writes the data using the MoR table type into the /moma_mor/artworks/ S3 object prefix. This example uses the BULK_INSERT write operation (op) and the filter-dupes option. The filter-dupes option ensures that should duplicate records from the source are dropped/filtered out before INSERT or BULK_INSERT. This set of options is ideal for the initial bulk inserting of existing data into Hudi tables. The job runs one time and completes, unlike the previous example that ran continuously.

	export DATA_LAKE_BUCKET="<your_data_lake_bucket_name>"

	# artworks data, MoR table type, 1x bulk insert
	spark-submit \
	–jars /usr/lib/spark/jars/spark-avro.jar,/usr/lib/hudi/hudi-utilities-bundle.jar \
	–conf spark.sql.catalogImplementation=hive \
	–class org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer /usr/lib/hudi/hudi-utilities-bundle.jar \
	–table-type MERGE_ON_READ \
	–source-ordering-field __source_ts_ms \
	–props "s3://${DATA_LAKE_BUCKET}/hudi/deltastreamer_artworks_apicurio_mor.properties" \
	–source-class org.apache.hudi.utilities.sources.AvroDFSSource \
	–target-base-path "s3://${DATA_LAKE_BUCKET}/moma/artworks_mor/" \
	–target-table moma_mor.artworks \
	–schemaprovider-class org.apache.hudi.utilities.schema.SchemaRegistryProvider \
	–enable-sync \
	–op BULK_INSERT \
	–filter-dupes

view raw deltastreamer_artists_1x.sh hosted with ❤ by GitHub

Syncing with Hive

The following abridged, video-only clip demonstrates the differences between the Hudi CoW and MoR table types with respect to Apache Hive. In the video, we run the deltastreamer_jobs_bulk_bkgd.sh script, included on GitHub. This script runs four different Apache Spark jobs, using Hudi DeltaStreamer to bulk-ingest all the artists and artworks CDC data from Amazon S3 into both Hudi CoW and MoR table types. Once the four Spark jobs are complete, the script queries Apache Hive and displays the new Hive databases and database tables created by DeltaStreamer.

Hudi DeltaStreamer Spark jobs running on the Amazon EMR

In both the video above and terminal screengrab below, note the difference in the tables created within the two Hive databases, the Hudi CoW table type (moma_cow) and the MoR table type (moma_mor). The MoR table type creates both a read-optimized table (_ro) as well as a real-time table (_rt) for each datasource (e.g., artists_ro and artists_rt).

View of the Apache Hive CoW and MoR database tables

According to documentation, Hudi creates two tables in the Hive metastore for the MoR table type. The first, a table which is a read-optimized view appended with _ro and the second, a table with the same name appended with _rt which is a real-time view. According to Hudi, the read-optimized view exposes columnar Parquet while the real-time view exposes columnar Parquet and/or row-based logs; you can query both tables. The CoW table type creates a single table without a suffix for each datasource (e.g., artists). Below, we see the Hive table structure for the artists_rt table, created by DeltaStreamer, using SHOW CREATE TABLE moma_mor.artists_rt;.

	CREATE EXTERNAL TABLE `moma_mor.artists_rt`(
	`_hoodie_commit_time` string,
	`_hoodie_commit_seqno` string,
	`_hoodie_record_key` string,
	`_hoodie_partition_path` string,
	`_hoodie_file_name` string,
	`artist_id` int,
	`name` string,
	`gender` string,
	`birth_year` int,
	`death_year` int,
	`__op` string,
	`__db` string,
	`__table` string,
	`__schema` string,
	`__lsn` bigint,
	`__source_ts_ms` bigint,
	`__deleted` string)
	PARTITIONED BY (
	`nationality` string)
	ROW FORMAT SERDE
	'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
	WITH SERDEPROPERTIES (
	'hoodie.query.as.ro.table'='false',
	'path'='s3://<your_data_lake_bucket>/moma/artists_mor')
	STORED AS INPUTFORMAT
	'org.apache.hudi.hadoop.realtime.HoodieParquetRealtimeInputFormat'
	OUTPUTFORMAT
	'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
	LOCATION
	's3://<your_data_lake_bucket>/moma/artists_mor'
	TBLPROPERTIES (
	'bucketing_version'='2',
	'last_commit_time_sync'='20211230180429',
	'spark.sql.partitionProvider'='catalog',
	'spark.sql.sources.provider'='hudi',
	'spark.sql.sources.schema.numPartCols'='1',
	'spark.sql.sources.schema.numParts'='1',
	'spark.sql.sources.schema.part.0'='{"type":"struct","fields":[{"name":"_hoodie_commit_time","type":"string","nullable":true,"metadata":{}},{"name":"_hoodie_commit_seqno","type":"string","nullable":true,"metadata":{}},{"name":"_hoodie_record_key","type":"string","nullable":true,"metadata":{}},{"name":"_hoodie_partition_path","type":"string","nullable":true,"metadata":{}},{"name":"_hoodie_file_name","type":"string","nullable":true,"metadata":{}},{"name":"artist_id","type":"integer","nullable":false,"metadata":{}},{"name":"name","type":"string","nullable":true,"metadata":{}},{"name":"gender","type":"string","nullable":true,"metadata":{}},{"name":"birth_year","type":"integer","nullable":true,"metadata":{}},{"name":"death_year","type":"integer","nullable":true,"metadata":{}},{"name":"__op","type":"string","nullable":true,"metadata":{}},{"name":"__db","type":"string","nullable":true,"metadata":{}},{"name":"__table","type":"string","nullable":true,"metadata":{}},{"name":"__schema","type":"string","nullable":true,"metadata":{}},{"name":"__lsn","type":"long","nullable":true,"metadata":{}},{"name":"__source_ts_ms","type":"long","nullable":true,"metadata":{}},{"name":"__deleted","type":"string","nullable":true,"metadata":{}},{"name":"nationality","type":"string","nullable":true,"metadata":{}}]}',
	'spark.sql.sources.schema.partCol.0'='nationality',
	'transient_lastDdlTime'='1640919578')

view raw moma_mor_artists_rt.sql hosted with ❤ by GitHub

Having run the demonstration’s deltastreamer_jobs_bulk_bkgd.sh script, the resulting object structure in the Hudi-managed section of the Amazon S3 bucket looks as follows.

S3 object structure in Hudi-managed Amazon S3 bucket

Below is an example of Hudi files created in the /moma/artists_cow/ S3 object prefix. When using data lake table formats like Hudi, given its specialized directory structure and the high number of objects, interactions with the data should be abstracted through Hudi’s programming interfaces. Generally speaking, you do not interact directly with the objects in a data lake.

	"moma/artists_cow/.hoodie/.aux/.bootstrap/.fileids_$folder$",
	"moma/artists_cow/.hoodie/.aux/.bootstrap/.partitions_$folder$",
	"moma/artists_cow/.hoodie/.aux/.bootstrap_$folder$",
	"moma/artists_cow/.hoodie/.aux_$folder$",
	"moma/artists_cow/.hoodie/.temp_$folder$",
	"moma/artists_cow/.hoodie/20211231203737.commit",
	"moma/artists_cow/.hoodie/20211231203737.commit.requested",
	"moma/artists_cow/.hoodie/20211231203737.inflight",
	"moma/artists_cow/.hoodie/20211231203738.rollback",
	"moma/artists_cow/.hoodie/20211231203738.rollback.inflight",
	"moma/artists_cow/.hoodie/archived_$folder$",
	"moma/artists_cow/.hoodie/hoodie.properties",
	"moma/artists_cow/.hoodie_$folder$",
	"moma/artists_cow/nationality=Afghan/.hoodie_partition_metadata",
	"moma/artists_cow/nationality=Afghan/4f39e019-e3d7-4a6a-a7bd-6d07eddd495a-0_0-28-3352_20211231203737.parquet",
	"moma/artists_cow/nationality=Afghan_$folder$",
	"moma/artists_cow/nationality=Albanian/.hoodie_partition_metadata",
	"moma/artists_cow/nationality=Albanian/4f39e019-e3d7-4a6a-a7bd-6d07eddd495a-1_0-28-3352_20211231203737.parquet",
	"moma/artists_cow/nationality=Albanian_$folder$",
	"moma/artists_cow/nationality=Algerian/.hoodie_partition_metadata",
	"moma/artists_cow/nationality=Algerian/4f39e019-e3d7-4a6a-a7bd-6d07eddd495a-2_0-28-3352_20211231203737.parquet",
	"moma/artists_cow/nationality=Algerian_$folder$",
	"moma/artists_cow/nationality=American/.hoodie_partition_metadata",
	"moma/artists_cow/nationality=American/0065ed77-4a6c-4755-b133-45126310936d-0_502-28-3854_20211231203737.parquet",
	"moma/artists_cow/nationality=American/011d5c57-c918-40d8-8518-c3cb56747133-0_15-28-3367_20211231203737.parquet"

view raw hudi_files_s3.txt hosted with ❤ by GitHub

Hudi CLI commands used in the next video

Hudi CLI

Optionally, we can inspect the Hudi tables using the Hudi CLI (hudi-cli). The CLI offers an extensive list of available commands. Using the CLI, we can inspect the Hudi tables and their schemas, and review operational statistics like write amplification (the number of bytes written for 1 byte of incoming data), commits, and compactions.

	> hudi-cli

	help
	connect –path s3://<your_data_lake_bucket>/moma/artworks_mor/
	connect –path s3://<your_data_lake_bucket>/moma/artworks_cow/
	desc
	fetch table schema
	commits show
	stats wa
	compactions show all

view raw hudi_cli_commands.txt hosted with ❤ by GitHub

Using the Hudi CLI from the Amazon EMR Master Node

The following short video-only clip shows the use of the Hudi CLI, running on the Amazon EMR Master Node, to inspect the Hudi tables in S3.

Using the Hudi CLI from the Amazon EMR Master Node

Hudi Data Structure

Recall the sample Kafka message we saw earlier in the post representing an insert of an artist record with the artist_id 1. Below, we see what the same record looks like after being ingested by Hudi DeltaStreamer. Note the five additional fields added by Hudi with the _hoodie_ prefix.

	{
	"_hoodie_commit_time": "20211227215352",
	"_hoodie_commit_seqno": "20211227215352_63_7301",
	"_hoodie_record_key": "1",
	"_hoodie_partition_path": "nationality=American",
	"_hoodie_file_name": "0e91bb5b-aa93-42a9-933d-242f5fda1b8f-0_63-24-4710_20211227215352.parquet",
	"artist_id": 1,
	"name": "Robert Arneson",
	"nationality": "American",
	"gender": "Male",
	"birth_year": 1930,
	"death_year": 1992,
	"__op": "r",
	"__db": "moma",
	"__table": "artists",
	"__schema": "public",
	"__lsn": 3637434647944,
	"__source_ts_ms": 1640566580452,
	"__deleted": "false"
	}

view raw final_hudi_record.json hosted with ❤ by GitHub

Querying Hudi-managed Data

With the initial data ingestion complete and the CDC and DeltaStreamer processes monitoring for future changes, we can query the resulting data stored in Hudi tables. First, we will make some changes to the PostgreSQL MoMA Collection database to see how Hudi manages the data mutations. We could also make changes directly to the Hudi tables using Hive, Spark, or Presto. However, that would cause our datasource to be out of sync with the Hudi tables, potentially negating the entire CDC process. When developing a data lake, this is a critically important consideration — how changes are introduced to Hudi tables, especially when CDC is involved, and whether data continuity between datasources and the data lake is essential.

For the demonstration, I have made a series of arbitrary updates to a piece of artwork in the MoMA Collection database, ‘Picador (La Pique)’ by Pablo Picasso.

	— 'Picador (La Pique)', by Pablo Picasso
	SELECT *
	FROM artworks
	WHERE artwork_id = 128447 AND classification = 'Print';

	— firts update (creation date)
	UPDATE artworks
	SET date = 1959
	WHERE artwork_id = 128447;

	— second update (acquisition date)
	UPDATE artworks
	SET acquisition_date = '2009-04-15'
	WHERE artwork_id = 128447;

	— third update (in vs. '')
	UPDATE artworks
	SET dimensions = 'composition: 20 13/16 x 25 3/16 in (52.9 x 64 cm); sheet: 24 7/16 x 29 1/2 in (62.1 x 75 cm)'
	WHERE artwork_id = 128447;

	— fourth update (acquisition date)
	UPDATE artworks
	SET acquisition_date = '2009-04-19'
	WHERE artwork_id = 128447;

view raw moma_hudi_changes.sql hosted with ❤ by GitHub

Below, note the last four objects shown in S3. Judging by the file names and dates, we can see that the CDC process, using Kafka Connect, has picked up the four updates I made to the record in the database. The Source Connector first wrote the changes to Kafka. The Sink Connector then read those Kafka messages and wrote the data to Amazon S3 in Avro format, as shown below.

Looking again at S3, we can also observe that DeltaStreamer picked up the new CDC objects in Amazon S3 and wrote them to both the Hudi CoW and MoR tables. Note the file types shown below. Given Hudi’s MoR table type structure, Hudi first logged the changes to row-based delta files and later compacted them to produce a new version of the columnar-format Parquet file.

Hudi MoR row-based delta log files and compacted Parquet files

Querying Results from Apache Hive

There are several ways to query Hudi-managed data in S3. In this demonstration, they include against Apache Hive using the hive client from the command line, against Hive using Spark, and against the Hudi tables also using Spark. We could also install Presto on EMR to query the Hudi data directly or via Hive.

Querying the real-time artwork_rt table in Hive after we make each database change, we can observe the data in Hudi reflects the updates. Note that the value of the _hoodie_file_name field for the first three updates is a Hudi delta log file, while the value for the last update is a Parquet file. The Parquet file signifies compaction occurred between the fourth update was made, and the time the Hive query was executed. Lastly, note the type of operation (_op) indicates an update change (u) for all records.

Querying the data in the Hudi MoR real-time table as we make changes to the database

Once all fours database updates are complete and compaction has occurred, we should observe identical results from all Hive tables. Below, note the _hoodie_file_name field for all three tables is a Parquet file. Logically, the Parquet file for the MoR read-optimized and real-time Hive tables is the same.

Querying the same record in all three Hive tables: Hudi MoR _ro and _rt tables and CoW table

Had we queried the data previous to compaction, the results would have differed. Below we have three queries. I further updated the artwork record, changing the date field from 1959 to 1960. The read-optimized MoR table, artworks_ro, still reflects the original date value, 1959, before the update and prior to compaction. The real-time table,artworks_rt , reflects the latest update to the date field, 1960. Note that the value of the _hoodie_file_name field for the read-optimized table is a Parquet file, while the value for the real-time table (artworks_rt), the third and final query, is a delta log file. The delta log allows the real-time table to display the most current state of the data in Hudi.

Querying the same record in all three Hive tables

Below are a few useful Hive commands to query the changes in Hudi.

	— beeline or hive

	— beeline connect
	!connect jdbc:hive2://localhost:10000/default

	SHOW DATABASES;
	DESCRIBE DATABASE moma_mor;

	USE moma_cow;SHOW TABLES;
	USE moma_mor;SHOW TABLES;
	USE moma_mor;DESCRIBE artworks_ro;

	MSCK REPAIR TABLE moma_mor.artworks_ro;
	SHOW PARTITIONS moma_mor.artworks_ro;

	ANALYZE TABLE moma_mor.artists_rt COMPUTE STATISTICS;
	DESCRIBE EXTENDED moma_mor.artists_rt;

	— test query performance without caching
	set hive.query.results.cache.enabled=false;

	— 100 rows selected (1.394 seconds) <- read-optimized vs. real-time table
	SELECT * FROM moma_mor.artworks_ro WHERE department='Prints & Illustrated Books' LIMIT 100;

	— 100 rows selected (2.371 seconds)
	SELECT * FROM moma_mor.artworks_rt WHERE department='Prints & Illustrated Books' LIMIT 100;

	— 10 rows selected (0.719 seconds) <- read-optimized vs. real-time table, classification is partitioned
	SELECT * FROM moma_mor.artworks_ro WHERE classification='Print' LIMIT 10;

	— 10 rows selected (1.482 seconds)
	SELECT * FROM moma_mor.artworks_rt WHERE classification='Print' LIMIT 10;

	EXPLAIN EXTENDED SELECT * FROM moma_mor.artworks_rt WHERE artwork_id=128447 AND classification='Print';

	— 1 row selected (14.126 seconds) <- read-optimized vs. real-time table
	SELECT * FROM moma_mor.artworks_ro WHERE artwork_id=128447;

	— 1 row selected (32.877 seconds)
	SELECT * FROM moma_mor.artworks_rt WHERE artwork_id=128447;

	— 1 row selected (1.491 seconds) <- classification is partitioned
	SELECT * FROM moma_mor.artworks_rt WHERE artwork_id=128447 AND classification='Print';

	— 84 rows selected (8.618 seconds)
	SELECT artworks.title AS title,
	artworks.`date` AS created,
	artworks.name AS artist,
	artists.nationality AS nationality,
	artworks.classification AS classification
	FROM moma_cow.artworks artworks
	JOIN moma_cow.artists artists ON (artworks.artist_id = artists.artist_id)
	WHERE artworks.artist_id = 4609
	AND nationality = 'Spanish'
	AND classification = 'Print'
	AND artworks.`date` IS NOT NULL
	ORDER BY created, title;

view raw hive_commands.hql hosted with ❤ by GitHub

Deletes with Hudi

In addition to inserts and updates (upserts), Apache Hudi can manage deletes. Hudi supports implementing two types of deletes on data stored in Hudi tables: soft deletes and hard deletes. Given this demonstration’s specific configuration for CDC and DeltaStreamer, we will use soft deletes. Soft deletes retain the record key and nullify the other field’s values. Hard deletes, a stronger form of deletion, physically remove any record trace from the Hudi table.

Below, we see the CDC record for the artist with artist_id 441. The event flattening single message transformation (SMT), used by the Debezium-based Kafka Connect Source Connector, adds the __deleted field with a value of true and nullifies all fields except the record’s key, artist_id, which is required.

	{
	"artist_id" : 441,
	"name" : null,
	"nationality" : null,
	"gender" : null,
	"birth_year" : null,
	"death_year" : null,
	"__op" : {
	"string" : "d"
	},
	"__db" : {
	"string" : "moma"
	},
	"__table" : {
	"string" : "artists"
	},
	"__schema" : {
	"string" : "public"
	},
	"__lsn" : {
	"long" : 3692866569488
	},
	"__source_ts_ms" : {
	"long" : 1640814436010
	},
	"__deleted" : {
	"string" : "true"
	}
	}

view raw cdc_delete_artist.json hosted with ❤ by GitHub

Below, we see the same delete record for the artist with artist_id 441 in the Hudi MoR table. All the null fields have been removed.

	{
	"_hoodie_commit_time": "20211229225047",
	"_hoodie_commit_seqno": "20211229225047_1_1",
	"_hoodie_record_key": "441",
	"_hoodie_partition_path": "nationality=default",
	"_hoodie_file_name": "2a98931a-6015-438e-be78-1eff80a75f83-2_1-24-15431_20211229225047.parquet",
	"artist_id": 441,
	"__op": "d",
	"__db": "moma",
	"__table": "artists",
	"__schema": "public",
	"__lsn": 3692866569488,
	"__source_ts_ms": 1640814436010,
	"__deleted": "true"
	}

view raw hudi_delete_artist.json hosted with ❤ by GitHub

Below, we see how the deleted record appears in the three Hive CoW and MoR artwork tables. Note the query results from the read-optimized MoR table, artworks_ro, contains two records — the original record (r) and the deleted record (d). The data is partitioned by nationality, and since the record was deleted, the nationality field is changed to null. In S3, Hudi represents this partition as nationality=default. The record now exists in two different Parquet files, within two separate partitions, something to be aware of when querying the read-optimized MoR table.

Results of a database delete as shown in Hive CoW and MoR tables

Time Travel

According to the documentation, Hudi has supported time travel queries since version 0.9.0. With time travel, you can query the previous state of your data. Time travel is particularly useful for use cases, including rollbacks, debugging, and audit history.

To demonstrate time travel queries in Hudi, we start by making some additional changes to the source database. For this demonstration, I made a series of five updates and finally a delete to the artist record with artist_id 299 in the PostgreSQL database over a few-hour period.

	— first update (birth)
	UPDATE public.artists
	SET birth_year = 1907
	WHERE artist_id = 299;

	— second update (death)
	UPDATE public.artists
	SET death_year = 1989
	WHERE artist_id = 299;

	— third update (middle initial)
	UPDATE public.artists
	SET name = 'Gerhard M. Bakker'
	WHERE artist_id = 299;

	— fourth update (nationality – impacts partitions)
	UPDATE public.artists
	SET nationality = 'German'
	WHERE artist_id = 299;

	— fifth update (birth)
	UPDATE public.artists
	SET birth_year = 1905
	WHERE artist_id = 299;

	— delete
	DELETE
	FROM public.artists
	WHERE artist_id = 299;

view raw hudi_time_travel.sql hosted with ❤ by GitHub

Once the CDC and DeltaStreamer ingestion processes are complete, we can use Hudi’s time travel query capability to view the state of data in Hudi at different points in time (instants). To do so, we need to provide an as.an.instant date/time value to Spark (see line 21 below).

Based on the time period in which I made the five updates and the delete, I have chosen six instants during that period where I want to examine the state of the record. Below is an example of the PySpark code from a Jupyter Notebook used to perform the six time travel queries against the Hudi MoR artist’s table.

	from datetime import timedelta
	from dateutil import parser

	base_path = "s3://open-data-lake-demo-us-east-1/moma/artists_mor"

	instances = [ # times in EST
	"2021-12-30 08:00:00", # reflects original record (r)
	"2021-12-30 09:00:00", # refects updates 1 and 2 (u)
	"2021-12-30 09:30:00", # refects updates 3 (u)
	"2021-12-30 11:00:00", # refects updates 4 (u)
	"2021-12-30 12:30:00", # refects updates 5 (u)
	"2021-12-30 14:00:00", # refects delete (d)
	]

	for instant in instants:
	as_of_instant = parser.parse(instant) + timedelta(hours=5) # adjust EST for UTC

	print(f"Record state as of: {as_of_instant}")

	artistsSnapshotDF = (
	spark.read.format("hudi").option("as.of.instant", as_of_instant).load(base_path)
	)

	artistsSnapshotDF.createOrReplaceTempView("hudi_artists_snapshot")

	spark.sql(
	"""
	SELECT _hoodie_commit_time, __op, _hoodie_partition_path, name, nationality, gender, birth_year, death_year
	FROM hudi_artists_snapshot
	WHERE artist_id=299;
	"""
	).show()

view raw spark_hudi_time_travel.py hosted with ❤ by GitHub

Below, we see the results of the time travel queries. At each instant, we can observe the mutating state of the data in the Hudi MoR Artist’s table, including the initial bulk insert of the existing snapshot of data (r) and the delete record (d). Since the delete made in the PostgreSQL database was recorded as a soft delete in Hudi, as opposed to a hard delete, we are still able to retrieve the record at any instant.

	Record state as of: 2021-12-30 13:00:00
	+——————-+—-+———————-+—————–+———–+——+———-+———-+
	\|_hoodie_commit_time\|__op\|_hoodie_partition_path\| name\|nationality\|gender\|birth_year\|death_year\|
	+——————-+—-+———————-+—————–+———–+——+———-+———-+
	\| 20211230034812\| r\| nationality=American\|Gerhard H. Bakker\| American\| Male\| 1906\| 1988\|
	+——————-+—-+———————-+—————–+———–+——+———-+———-+

	Record state as of: 2021-12-30 14:00:00
	+——————-+—-+———————-+—————–+———–+——+———-+———-+
	\|_hoodie_commit_time\|__op\|_hoodie_partition_path\| name\|nationality\|gender\|birth_year\|death_year\|
	+——————-+—-+———————-+—————–+———–+——+———-+———-+
	\| 20211230132628\| u\| nationality=American\|Gerhard H. Bakker\| American\| Male\| 1907\| 1989\|
	+——————-+—-+———————-+—————–+———–+——+———-+———-+

	Record state as of: 2021-12-30 14:30:00
	+——————-+—-+———————-+—————–+———–+——+———-+———-+
	\|_hoodie_commit_time\|__op\|_hoodie_partition_path\| name\|nationality\|gender\|birth_year\|death_year\|
	+——————-+—-+———————-+—————–+———–+——+———-+———-+
	\| 20211230142035\| u\| nationality=American\|Gerhard M. Bakker\| American\| Male\| 1907\| 1989\|
	+——————-+—-+———————-+—————–+———–+——+———-+———-+

	Record state as of: 2021-12-30 16:00:00
	+——————-+—-+———————-+—————–+———–+——+———-+———-+
	\|_hoodie_commit_time\|__op\|_hoodie_partition_path\| name\|nationality\|gender\|birth_year\|death_year\|
	+——————-+—-+———————-+—————–+———–+——+———-+———-+
	\| 20211230144237\| u\| nationality=German\|Gerhard M. Bakker\| German\| Male\| 1907\| 1989\|
	+——————-+—-+———————-+—————–+———–+——+———-+———-+

	Record state as of: 2021-12-30 17:30:00
	+——————-+—-+———————-+—————–+———–+——+———-+———-+
	\|_hoodie_commit_time\|__op\|_hoodie_partition_path\| name\|nationality\|gender\|birth_year\|death_year\|
	+——————-+—-+———————-+—————–+———–+——+———-+———-+
	\| 20211230171925\| u\| nationality=German\|Gerhard M. Bakker\| German\| Male\| 1905\| 1989\|
	+——————-+—-+———————-+—————–+———–+——+———-+———-+

	Record state as of: 2021-12-30 19:00:00
	+——————-+—-+———————-+—————–+———–+——+———-+———-+
	\|_hoodie_commit_time\|__op\|_hoodie_partition_path\| name\|nationality\|gender\|birth_year\|death_year\|
	+——————-+—-+———————-+—————–+———–+——+———-+———-+
	\| 20211230180429\| d\| nationality=default\| null\| null\| null\| null\| null\|
	+——————-+—-+———————-+—————–+———–+——+———-+———-+

view raw hudi_time_travel_artist.txt hosted with ❤ by GitHub

Results of the time travel queries, ordered by commit time

In addition to time travel queries, Hudi also offers incremental queries and point in time queries.

Conclusion

Although this post only scratches the surface of the capabilities of Debezium and Hudi, you can see the power of CDC using Kafka Connect and Debezium, combined with Hudi, to build and manage open data lakes on AWS.

This blog represents my own viewpoints and not of my employer, Amazon Web Services (AWS). All product names, logos, and brands are the property of their respective owners.

Apache Hive, Apache Hudi, Apache Kafka, AWS, CDC, Change Data Capture, Data Lake, DeltaStreamer

Leave a comment

Video Demonstration: Building a Data Lake with Apache Airflow

Posted by Gary A. Stafford in Analytics, AWS, Big Data, Build Automation, Cloud, Python on November 12, 2021

Build a simple Data Lake on AWS using a combination of services, including Amazon Managed Workflows for Apache Airflow (Amazon MWAA), AWS Glue, AWS Glue Studio, Amazon Athena, and Amazon S3

Introduction

In the following video demonstration, we will build a simple data lake on AWS using a combination of services, including Amazon Managed Workflows for Apache Airflow (Amazon MWAA), AWS Glue Data Catalog, AWS Glue Crawlers, AWS Glue Jobs, AWS Glue Studio, Amazon Athena, Amazon Relational Database Service (Amazon RDS), and Amazon S3.

Using a series of Airflow DAGs (Directed Acyclic Graphs), we will catalog and move data from three separate data sources into our Amazon S3-based data lake. Once in the data lake, we will perform ETL (or more accurately ELT) on the raw data — cleansing, augmenting, and preparing it for data analytics. Finally, we will perform aggregations on the refined data and write those final datasets back to our data lake. The data lake will be organized around the data lake pattern of bronze (aka raw), silver (aka refined), and gold (aka aggregated) data, popularized by Databricks.

Architecture and workflow demonstrated in the video

Demonstration

For best results, view at 1080p HD on YouTube

Source Code

The source code for this demonstration, including the Airflow DAGs, SQL files, and data files, is open-sourced and located on GitHub.

DAGs

The DAGs shown in the video demonstration have been renamed for easier project management within the Airflow UI. The DAGs included in the GitHub project are as follows:

This blog represents my own viewpoints and not of my employer, Amazon Web Services (AWS). All product names, logos, and brands are the property of their respective owners.

Amazon Athena, Analytics, Apache Airflow, AWS, AWS Glue, Data Lake

Leave a comment

Video Demonstration: Building a Data Lake on AWS

Posted by Gary A. Stafford in Analytics, AWS, Big Data, Cloud, Python on November 12, 2021

Build a simple Data Lake on AWS using a combination of services, including AWS Glue, AWS Glue Studio, Amazon Athena, and Amazon S3

Introduction

In the following video demonstration, we will build a simple data lake on AWS using a combination of services, including AWS Glue Data Catalog, AWS Glue Crawlers, AWS Glue Jobs, AWS Glue Studio, Amazon Athena, Amazon Relational Database Service (Amazon RDS), and Amazon S3.

We will catalog and move data from three separate data sources into our Amazon S3-based data lake. Once in the data lake, we will perform ETL (or more accurately ELT) on the raw data — cleansing, augmenting, and preparing it for data analytics. Finally, we will perform aggregations on the refined data and write those final datasets back to our data lake. The data lake will be organized around the data lake pattern of bronze (aka raw), silver (aka refined), and gold (aka aggregated) data, popularized by Databricks.

Demonstration

For best results, view at 1080p HD on YouTube

Source Code

The source code for this demonstration, including the SQL statements, is open-sourced and located on GitHub.

This blog represents my own viewpoints and not of my employer, Amazon Web Services (AWS). All product names, logos, and brands are the property of their respective owners.

Amazon Athena, Analytics, AWS, AWS Glue, Data Lake

Leave a comment

Employing Amazon Macie to Discover and Protect Sensitive Data in your Amazon S3-based Data Lake

Posted by Gary A. Stafford in AWS, Bash Scripting, Big Data, Cloud, Enterprise Software Development, Python on March 15, 2021

Introduction

Working with Analytics customers, it’s not uncommon to see data lakes with a dozen or more discrete data sources. Data typically originates from sources both internal and external to the customer. Internal data may come from multiple teams, departments, divisions, and enterprise systems. External data comes from vendors, partners, public sources, and subscriptions to licensed data sources. The volume, velocity, variety, veracity, and method of delivery vary across the data sources. All this data is being fed into data lakes for purposes such as analytics, business intelligence, and machine learning.

Given the growing volumes of incoming data and variations amongst data sources, it is increasingly complex, expensive, and time-consuming for organizations to ensure compliance with relevant laws, policies, and regulations. Regulations that impact how data is handled in a data lake include the Organizations Health Insurance Portability and Accountability Act (HIPAA), General Data Privacy Regulation (GDPR), Payment Card Industry Data Security Standard (PCI DSS), California Consumer Privacy Act (CCPA), and the Federal Information Security Management Act (FISMA).

Data Lake

AWS defines a data lake as a centralized repository that allows you to store all your structured and unstructured data at any scale. Once in the data lake, you run different types of analytics — from dashboards and visualizations to big data processing, real-time analytics, and machine learning to guide better decisions.

Data in a data lake is regularly organized or separated by its stage in the analytics process. Incoming data is often referred to as raw data. Data is then processed — cleansed, filtered, enriched, and tokenized if necessary. Lastly, the data is analyzed and aggregated, and the results are written back to the data lake. The analyzed and aggregated data is used to build business intelligence dashboards and reports, machine learning models, and is delivered to downstream or external systems. The different categories of data — raw, processed, and aggregated, are frequently referred to as bronze, silver, and gold, a reference to their overall data quality or value.

Protecting the Data Lake

Imagine you’ve received a large volume of data from an external data source. The incoming data is cleansed, filtered, and enriched. The data is re-formatted, partitioned, compressed for analytical efficiency, and written back to the data lake. Your analytics pipelines run complex and time-consuming queries against the data. Unfortunately, while building reports for a set of stakeholders, you realize that the original data accidentally included credit card information and other sensitive information about your customers. In addition to being out of compliance, you have the wasted time and expense of the initial data processing, as well as the extra time and expense to replace and re-process the data. The solution — Amazon Macie.

Amazon Macie

According to AWS, Amazon Macie is a fully managed data security and data privacy service that uses machine learning and pattern matching to discover and protect your sensitive data stored in Amazon Simple Storage Service (Amazon S3). Macie’s alerts, or findings, can be searched, filtered, and sent to Amazon EventBridge, formerly called Amazon CloudWatch Events, for easy integration with existing workflow or event management systems, or to be used in combination with AWS services, such as AWS Step Functions or Amazon Managed Workflows for Apache Airflow (MWAA) to take automated remediation actions.

Data Discovery and Protection

In this post, we will deploy an automated data inspection workflow to examine sample data in an S3-based data lake. Amazon Macie will examine data files uploaded to an encrypted S3 bucket. If sensitive data is discovered within the files, the files will be moved to an encrypted isolation bucket for further investigation. Email and SMS text alerts will be sent. This workflow will leverage Amazon EventBridge, Amazon Simple Notification Service (Amazon SNS), AWS Lambda, and AWS Systems Manager Parameter Store.

Macie data inspection workflow architecture

Source Code

Using this git clone command, download a copy of this post’s GitHub repository to your local environment.

git clone --branch main --single-branch --depth 1 --no-tags \
    https://github.com/garystafford/macie-demo.git

AWS resources for this post can be deployed using AWS CloudFormation. To follow along, you will need recent versions of Python 3,

Programmatic Ponderings

Posts Tagged Data Lake

Utilizing In-memory Data Caching to Enhance the Performance of Data Lake-based Applications

Introduction

Caching for Data Lakes

In-memory Caching

Redis In-memory Data Store

Amazon ElastiCache for Redis

ElastiCache Performance Results

Source Code

Code Changes

Gradle Build

Application Properties

Athena Application Class

POJO Data Model Classes

Service Classes

Controller Classes

Amazon ElastiCache for Redis

Testing the Cache

Conclusion

Developing Spring Boot Applications for Querying Data Lakes on AWS using Amazon Athena

Introduction

TL;DR

Data Lake Data Source

Spring Projects with IntelliJ IDE

Visual Studio Code

Gradle

Amazon Corretto

Source Code

Amazon Athena Access

POJO-based Data Model Class

Service Class

Controller Class

Querying Views

Executing Prepared Statements

Working with Named Queries

Configuration

AWS IAM: Authentication and Authorization

Swagger UI and the OpenAPI Specification

Integration Tests

Load Testing the Service

Metrics

Conclusion

The Art of Building Open Data Lakes with Apache Hudi, Kafka, Hive, and Debezium

Introduction

Workflow

Alternatives

Source Code

Kafka Connect

Source Connector

Sink Connector

Running Kafka Connect

Apache Hudi

DeltaStreamer

Amazon EMR

DeltaStreamer Configuration

Running DeltaStreamer

Syncing with Hive

Hudi CLI

Hudi Data Structure

Querying Hudi-managed Data

Querying Results from Apache Hive

Deletes with Hudi

Time Travel

Conclusion

Video Demonstration: Building a Data Lake with Apache Airflow

Introduction

Demonstration

Source Code

DAGs

Video Demonstration: Building a Data Lake on AWS

Introduction

Demonstration

Source Code

Employing Amazon Macie to Discover and Protect Sensitive Data in your Amazon S3-based Data Lake

Introduction

Data Lake

Protecting the Data Lake

Amazon Macie

Data Discovery and Protection