Mastering millisecond latency and millions of events: The event-driven architecture behind the Amazon Key Suite

Post Syndicated from Ali Ufuk Yucel original https://aws.amazon.com/blogs/architecture/mastering-millisecond-latency-and-millions-of-events-the-event-driven-architecture-behind-the-amazon-key-suite/

Background

Amazon Key empowers customers to securely manage access to their homes and businesses through innovative solutions. Through a suite of consumer and business products, the Amazon Key team is transforming how customers receive deliveries and manage access to their spaces. Our In-Garage Delivery service offers a secure and convenient solution for receiving Amazon packages and groceries directly inside customers’ garages. For property managers and building owners, Amazon Key provides comprehensive access management solutions that enable safe and efficient delivery operations in apartment buildings and gated communities, enhancing both security and convenience for residents.

In this post, we explore how the Amazon Key team used Amazon EventBridge to modernize their architecture, transforming a tightly coupled monolithic system into a resilient, event-driven solution. We explore the technical challenges we faced, our implementation approach, and the architectural patterns that helped us achieve improved reliability and scalability. The post covers our solutions for managing event schemas at scale, handling multiple service integrations efficiently, and building an extensible architecture that accommodates future growth.

Opportunities

Service Coupling and System Fragility

Our legacy architecture faced significant challenges stemming from its tightly coupled design, where service interactions created a complex web of dependencies impacting system stability and scalability. Making service modifications was particularly challenging, as adding or removing services required careful consideration of numerous interdependencies. An incident highlighted this vulnerability when an issue in Service-A triggered a cascade of failures across many upstream services, with increased timeouts leading to retry attempts and ultimately resulting in service deadlocks. System fragility was further demonstrated when problems with a single device vendor, despite being responsible only for specific delivery operations, caused widespread degradation across multiple system services.

Loose Event Schemas

Our old event management infrastructure lacked explicit schema definitions and employed a loosely-typed data architecture, leading to several critical issues. Events were difficult to maintain as use cases expanded, and the absence of formal schema documentation impacted transparency and team collaboration. The design made it almost impossible to implement backward-incompatible changes, such as removing unused fields or events for performance optimization. Without a repository for schema management, team-to-team collaboration for schema modifications (adding fields, removing fields, deprecating fields, or marking fields as required) became challenging. The system also lacked organized validation logic, making it difficult for publishers to identify invalid events before they entered the system. Additionally, the loosely typed schemas lost important semantic context, such as inheritance and composition relationships between different event schemas.

Inconsistent Event Routing and Management

The event routing logic was manually managed and lacked the sophistication needed for growing use cases. The system only supported basic validation of events, primarily checking for required fields, with limited capability for extending validation rules or implementing more complex routing logic. Features that were commonly available in off-the-shelf solutions, such as parallel publishing to multiple subscribers, required significant custom development and ongoing maintenance effort. The implementation only supported a limited number of subscribers to the event pipeline, with no sustainable pathway for adding more consumers. While attempts were made to reduce coupling through SNS/SQS pairs between services, these solutions were implemented on an ad-hoc basis, lacking standardization and creating additional maintenance overhead. This approach led to redundant work and failed to abstract away common functionality, resulting in an inefficient and hard-to-maintain system.These challenges collectively highlighted the need for a more robust and flexible architectural approach that could better serve the system’s evolving needs while improving reliability, maintainability, and scalability.

Design

Given our requirements and the architectural challenges we faced, we implemented a single-bus, multi-account pattern to optimize our system architecture. In this design, each service team maintains complete ownership and autonomy over their application stack, enabling independent development and deployment cycles. Meanwhile, our DevOps team manages a centralized infrastructure stack that encompasses event bus rules, target configurations, and service integrations. This separation of concerns provides several key benefits:

  1. Clear ownership boundaries: Service teams can focus on their core business logic while leveraging a standardized event infrastructure.
  2. Centralized governance: The DevOps team facilitates consistent event routing patterns, security controls, and monitoring across service integrations.
  3. Simplified operations: A single event bus reduces operational complexity while maintaining logical separation through well-defined routing rules.
  4. Enhanced security: The multi-account structure provides natural isolation boundaries while still enabling controlled cross-account event flows.
  5. Streamlined compliance: Centralized management of data exchange patterns makes it easier to implement and maintain compliance requirements.

While EventBridge provided the foundation, we developed additional components to meet our specific requirements.  Our team built three key components: a schema repository serving as the single source of truth for event definitions, a client library that handles schema validation and provides developer-friendly abstractions, and an infrastructure library offering reusable components for subscriber integration.

Event Schema Repository

Amazon EventBridge’s schema discovery and documentation capabilities provide powerful solutions for managing event-driven architectures. The service automatically captures event structures in the schema registry, maintaining versions as events evolve over time. While EventBridge provides developers with tools to implement validation using external solutions or custom application code, it currently does not include native schema validation capabilities. For our organization’s large-scale event-driven architecture, schema validation was a critical requirement. We evaluated two implementation approaches: a centralized validation service or client-side validation at the publisher/subscriber level. The centralized approach would have required managing additional infrastructure, scaling considerations, and introduced latency through extra network hops. After analyzing these factors alongside our requirements for schema governance and team autonomy, we implemented a custom schema repository with client-side validation.

This architecture prioritizes developer experience through immediate validation feedback while maintaining our standards for schema versioning and release management. The repository serves as the foundation for our event-driven architecture, providing essential capabilities for data governance and quality control. By acting as the single source of truth for event definitions, it enables standardized validation across clients, enforces data quality checks, establishes clear ownership boundaries, and maintains comprehensive audit trails for schema changes. Publishers and subscribers leverage these schemas to maintain data consistency and compatibility as their services evolve. The repository has become instrumental in facilitating efficient cross-team collaboration through self-service schema discovery, documentation, and automated validation during development. It maintains a comprehensive registry of event publishers and their corresponding subscribers, providing clear visibility into event flow patterns and dependencies across the system. Teams can quickly manage schema evolution with clear deprecation policies and migration paths, while the system helps detect breaking changes early in the development cycle. This collaborative approach has significantly improved team velocity and reduced integration issues between services.

{
    "$schema": "http://json-schema.org/draft-04/schema#",
    "$id": "/resource/event/schema/EventV1.json",
    "title": "EventV1",
    "description": "Schema for a simple event.",
    "type": "object",
    "properties": {
        "id": {
            "description": "Id of the event.",
            "type": "string"
        },
        "type": {
            "description": "Type of the event.",
            "$ref": "EventType.json"
        },
        "time": {
            "description": "Time at which the event occurred. It uses ISO 8601 Date Time Format. Reference: https://www.iso.org/iso-8601-date-and-time-format.html",
            "type": "string",
            "format": "date-time"
        },
        "publisher": {
            "description": "Publisher of the event.",
            "$ref": "../core/Publisher.json"
        }
    },
    "required": [
        "id",
        "type",
        "time",
        "publisher"
    ]
}

Client Library

The client library serves as a crucial component for both publishers and subscribers, streamlining their integration with the central event bus. At its core, the library leverages our Event Schema Repository, generating code bindings at build time to provide developers with type-safe and intuitive interfaces for event creation and handling. This approach significantly enhances developer productivity by offering straightforward and convenient methods to construct events and interact with the bus, reducing the likelihood of errors and improving code readability.

A key feature of the client library is its built-in validation mechanism. By utilizing the schemas from our local repository, the library performs thorough validation of events before they are published. This proactive approach catches potential issues early in the development cycle, making sure that only well-formed events conforming to the agreed-upon schemas make it to the event bus. Once validated, the library handles the serialization process and manages the actual publishing of events to the bus, abstracting and simplifying data transformation and transport.

For subscribers, the client library offers equally valuable functionality. It seamlessly handles the deserialization of incoming events, presenting them to the subscribing services in a readily usable format. This feature saves development time and reduces the risk of parsing errors, allowing teams to focus on business logic rather than data handling intricacies. By providing these comprehensive capabilities, our client library has become an indispensable tool in our event-driven network, promoting consistency, reliability, and efficiency across our microservices architecture.

Subscriber Constructs Library

We developed a subscriber constructs library using AWS Cloud Development Kit (CDK) to simplify and standardize the integration process with our central event bus. This library abstracts the setup and management of underlying infrastructure required for event consumption, enabling teams to focus on their core business logic rather than infrastructure configuration details.

The library automates the creation of essential components required for reliable event processing. It provisions a dedicated event bus within the subscriber’s account, establishes the necessary IAM roles and permissions for secure cross-account communication with the central event bus, and configures standardized monitoring and alerting for event processing. This automation not only reduces the potential for configuration errors but also facilitates consistent implementation of our architectural patterns across different teams.

/**
 * Subscriber implementation to provision necessary AWS infrastructure.
 *
 */
const subscription = new Subscription(scope, id, {
    name: "DeliveryService", // Name of your application
    application: {
       region: Region.US_EAST_1, // Region of your Application
    },
});

Conclusion

Amazon Key team’s journey to modernize their architecture and build a resilient, event-driven solution exemplifies the powerful benefits of leveraging AWS EventBridge and adopting a well-designed event-driven architecture. By addressing the challenges of service coupling, loose event schemas, and inconsistent event routing, the team was able to transform their system into a more reliable, scalable, and maintainable resource. The key architectural patterns and components they implemented have had a significant impact on their ability to deliver innovative solutions to their customers.

Reliability and Scale:

  • Built a decoupled event system processing 2000 events/second with 99.99% success rate
  • Achieved consistent 80ms p90 latency from ingestion to target invocation across 14M subscriber calls
  • Avoided the need for new infrastructure for event exchange through standardized event routing
  • Enabled migration of existing complex interdependencies to event-driven architecture

Developer Experience:

  • Reduced service integration time for new use cases from five days to one day (80% improvement)
  • New event onboarding on the Custom Event Schema repository now takes four hours, down from 48 hours
  • Publisher/subscriber integration completed in eight hours, previously took 40 hours
  • Standardized client library addressed 90% of common integration errors

Security and Governance :

  • Single control plane manages 100% of event bus infrastructure
  • Automated security compliance checks catch 100% of unauthorized data exchange patterns
  • Real-time monitoring dashboard tracks every event flow and schema change
  • Schema repository provides complete audit trail for system modifications

The solutions developed by the Amazon Key team provide a blueprint for other organizations looking to modernize their architectures and leverage the power of event-driven design patterns. By adopting similar architectural patterns and components, such as the schema repository and client libraries, other organizations can be empowered to achieve similar benefits.

About the authors

[$] Sigil simplifies creating and editing EPUBs

Post Syndicated from jzb original https://lwn.net/Articles/1054751/

Creating an ebook in EPUB format is easy,
for certain values of “easy”. All one really needs
is a text editor, a few command-line utilities; also needed is a working
knowledge of XHTML, CSS, along with an understanding of the format’s
structure and required boilerplate. Creating
a well-formatted and attractive ebook is a bit harder. However, it can be
made easier with an application custom-made for the purpose. Sigil is an EPUB editor that
provides the tooling authors and publishers may be looking for.

LibreOffice 26.2 released

Post Syndicated from jake original https://lwn.net/Articles/1057256/

Version 26.2 of the LibreOffice
office suite has been released.

LibreOffice 26.2 is focused on improvements that make a difference in daily work and brings better performance, smoother interaction with complex documents and improved compatibility with files created in other office software. Whether you’re writing reports, managing spreadsheets, or preparing presentations, the experience feels more responsive and reliable.

LibreOffice has always been about giving users control. LibreOffice 26.2 continues that tradition by strengthening support for open document standards, and ensuring long-term access to your files, without subscriptions, license restrictions, or data collection. Your documents stay yours – forever.

More information can be found in the release notes
for LibreOffice 26.2.

Security updates for Wednesday

Post Syndicated from jake original https://lwn.net/Articles/1057247/

Security updates have been issued by Debian (thunderbird), Fedora (openqa, os-autoinst, python-jupytext, python-python-multipart, rust-sequoia-keystore-server, rust-sequoia-octopus-librnp, rust-sequoia-sq, rust-sequoia-sqv, and xen), Oracle (curl, kernel, net-snmp, python3, and python3.12), Red Hat (container-tools:rhel8, fence-agents, golang, golang-github-openprinting-ipp-usb, grafana, grafana-pcp, opentelemetry-collector, podman, python-s3transfer, python-wheel, and resource-agents), SUSE (alloy, chromium, cockpit-podman, cockpit-subscriptions, dpdk, elemental-register, elemental-toolkit, glib2, glibc, gpg2, ImageMagick, imagemagick, jasper, java-17-openjdk, java-21-openjdk, kernel, libheif, libmlt++, libpng16, libsodium, libsoup, libvirt, openssl-3, openvpn, php8, postgresql16, postgresql17 and postgresql18, protobuf, python-FontTools, python-fonttools, python-h2, python-python-multipart, python-urllib3, python-wheel, python311-PyNaCl, trivy, ucode-amd, udisks2, unbound, util-linux, wireshark, and xkbcomp), and Ubuntu (emacs, freerdp2, glibc, imagemagick, mysql-8.0, pagure, python-django, python-filelock, python-internetarchive, and python-keystonemiddleware).

Kelly Hiscoe Recognized Among CRN 2026 Channel Chiefs for Innovation and Impact

Post Syndicated from Rapid7 original https://www.rapid7.com/blog/post/c-kelly-hiscoe-recognized-crn-2026-channel-chiefs-innovation-impact

In 2026, security teams are still grappling with the challenges posed by expanding attack surfaces and persistent resource constraints. Together with the rapid onset of AI-driven threats, security leaders are weathering this ‘perfect storm’ by seeking consolidation of their technology stacks – favoring trusted partnerships that truly understand their unique ecosystems.

To elevate security partners from mere service providers to essential, trusted security advisors, it is vital to help customers achieve a comprehensive view of their IT environments. This includes a clear understanding of their risk profiles and a cohesive approach to continuous detection, response, and compliance, says Kelly Hiscoe, Sr. Director, Global Partner Programs & Experience.

Kelly brings to Rapid7 more than 17 years of experience in cybersecurity channel ecosystems. And after being named to CRN’s Women of the Channel list in 2020, 2024, and 2025, CRN has honored her as a Channel Chief for 2026.

She has consistently led her teams to design competitive programs, drive operational excellence, and enhance the partner experience from the ground-up. Here’s what makes her tick, and how Kelly (and Rapid7) are thinking about the channel in 2026 and beyond.

A channel philosophy rooted in shared responsibility

Kelly’s approach to the channel is grounded in the simple belief that true success is built through shared ownership. Rather than being confined to a single team, channel success must be woven into a company’s DNA: reflected in its processes, tools, and most importantly, how sales teams consistently engage with partners.

For Kelly, this means a company-wide commitment to engaging and collaborating with partners, in a way that, at its heart, exists to help customers achieve their goals. That’s what “Channel-first” means, and what Rapid7 aims to reflect.

Refreshing Rapid7’s partner ecosystem

In February 2025, Rapid7 launched its reimagined PACT Partner Program, and Kelly led the global team responsible for that launch. The revamped program was designed to equip partners with the tools, training, and resources needed to address evolving global security challenges together. 

Key enhancements included a modernized Partner Portal that enables real-time collaboration and automation, as well as tailored engagement programs and specializations, plus the launch of the Rapid7 Partner Academy. Since its debut, the Academy has seen more than 2,000 partner learners earn over 3,700 certifications. Rapid7’s partners consistently highlight its clarity, relevance, and impact in deepening cybersecurity expertise.

Looking ahead: Helping partners navigate 2026

As consolidation continues and competition in the market grows, partners are facing more challenges than ever in navigating that complexity and standing out amid the noise. Kelly remains focused on helping partners align with vendors that deliver clear, customer-centric value, comprehensive coverage across the expanding attack surface, and predictable engagement models. You can read Kelly’s full CRN Channel Chief details here.

US Declassifies Information on JUMPSEAT Spy Satellites

Post Syndicated from Bruce Schneier original https://www.schneier.com/blog/archives/2026/02/us-declassifies-information-on-jumpseat-spy-satellites.html

The US National Reconnaissance Office has declassified information about a fleet of spy satellites operating between 1971 and 2006.

I’m actually impressed to see a declassification only two decades after decommission.

Обединена Европа и европейската мечта

Post Syndicated from Искрен Иванов original https://www.toest.bg/obedinena-evropa-i-evropeyskata-mechta/

Световният мир не може да бъде запазен без съзидателни усилия, пропорционални на опасностите, които го грозят.

Обединена Европа и европейската мечта

Тези думи произнася френският външен министър Роберт Шуман на 9 май 1950 г. – момента, в който се ражда идеята за Европейската общност за въглища и стомана. Краят на Втората световна война бележи края на европейските илюзии, че световният мир може да бъде постигнат, без да съществуват правила, с които той да се опазва.

Бившите колониални империи, владели векове наред глобалните ресурси и наложили правото на силата по всички континенти, са поставени на колене между двете суперсили – САЩ, които се превръщат в глобален кредитор на целия свят, и СССР, който разполага с близо тримилионна армия, готова да окупира останалата част от Стария континент. 

Неслучайно в декларацията си Шуман поставя две условия за обединението на Европа: икономическата интеграция, с която най-сетне да се сложи край на френско-германското съперничество на континента, и подхода step-by-step, който предполага постепенно интегриране на европейските държави в едно цяло. 

Днес Европа отново е обединена, но някак си безсилна да се противопостави на войните, които Великите сили водят, за да придобият глобално надмощие. Парадоксално, но не неочаквано, по-голямата част от човечеството избра да живее отново в условията на надпревара, а не на мир.

Какво стана с европейската стратегическа култура? 

Стратегическото мислене на Европа винаги е било имперско – факт, видим както от историята на нашия континент, така и от стремежа му да се обедини. Проявленията на тази стратегическа култура придобиват различни форми, като се започне от Римската империя и нейната концепция за Mare Nostrum, мине се през средновековната идея за Imperium Christianum и се стигне до стремежа на европейските владетели да обединят Европа под своята корона. 

Обединена Европа и европейската мечта
Римската империя по времето на Траян, когато достига най-голямото си териториално разширение и когато Средиземно море се явява вътрешно за нея (Mare Nostrum в буквален превод означава „наше море“). Почти като нашите три морета, но не съвсем. Източник: Wikimedia Commons / Tataryn

По правило такъв тип геополитическо мислене неизбежно води до свят, в който войната се превръща в основен инструмент на държавната политика. Вестфалският мирен договор, Виенският конгрес и Версайско-Вашингтонската система от споразумения така и не успяват да победят европейския порив към власт, в който сякаш дреме сянката на древногръцкия стремеж към колонизация и опознаване на останалия свят.

Какво се промени след самоубийствената Втора световна война в Европа? Не, причината Европа да загуби бойния си дух не е само в разделението на Германия и поставянето на ресурсите ѝ под контрола на Европейската общност за въглища и стомана. Тя е в преосмислянето на имперската идея като цяло. 

Идеологическият флаг на този преломен момент може да бъде открит не в Декларацията на Шуман, а в творчеството на Макс Хоркхаймер и Теодор Адорно, които полагат основите на диалектиката на Просвещението. Тази система от идеи цели да противодейства на старите доктрини за създаването на Свещена римска империя или за възстановяването на Древния Рим, тъй като Хоркхаймер и Адорно виждат в тях причината за възхода на националсоциализма и расистката му политика. 

В следващите десетилетия диалектиката на Просвещението постепенно залива Европа, формирайки европейската интеграция като светски, напълно автономен от религията процес, чиято крайна цел е икономическото, а не политическото обединение на Общността. Крайната цел се състои в окончателното лишаване на западните общества от имперското им мислене и провеждането на мирна културна революция, която да промени стратегическата им култура завинаги.

В САЩ тези идеи не успяват да пробият до края на Студената война поради силното пуританско крило в американската външна политика и необходимостта от сдържането на СССР. В Европа обаче те проникват изключително лесно, което кара европейските елити да предлагат смели проекти за политическо обединение, които така и не биват реализирани. Пример в това отношение е опитът на държавите членки да създадат Европейско политическо сътрудничество, което в крайна сметка става факт през 70-те години на миналия век, но се свежда единствено до създаването на общи механизми за неформални консултации в Европейската икономическа общност. 

Но дори когато европейският проект е спасен благодарение на Единния европейски акт и възкръсва като Европейски съюз с Договора от Маастрихт през 1992 г., гласовете на политиците, които повдигат въпроса за създаването на европейска армия, биват заглушени заради опасенията от завръщането на европейския национализъм. 

Така след края на Студената война европейската стратегическа култура окончателно се разпада на отделни национални единици, които въпреки разширяването на ЕС все повече се отдалечават от завета на европейските бащи основатели.

Обединена Европа и европейската мечта
Договорът от Маастрихт, който е подписан на 7 февруари 1992 г. и поставя началото на Европейския съюз. Източник: Wikimedia Commons / Museum der Bayerischen Geschichte, CC BY 2.0 DE

Възможно ли е европейската стратегическа култура да бъде съживена?

Разпадът на европейската стратегическа култура поради невъзможността на Европа да създаде собствена армия е сходен с началото на упадъка в хегемонията на САЩ, които след края на Студената война постепенно губят усещането къде са границите на тяхната сила. Изчезването на СССР от картата на света и отпадането на необходимостта от идеологическо сдържане превръща Запада в най-силният актьор в системата на международните отношения, като същевременно полага основите на неговия упадък. 

Причината е, че когато нямаш срещу какво да се бориш, лесно се увличаш от собствената си възможност да проектираш сила в глобален план. В този смисъл имперската метафора е най-неподходящият начин да опишем САЩ и Европа след разпада на СССР. Империите не проектират сила просто като я налагат, но и като конвертират нагласите на завладените територии, асимилирайки културата им. 

Все едно да кажеш, че за няколко години българите станаха толерантни европейци по манталитет или пък предприемчиви, работещи хора, подобно на американците. В много отношения изкуственото създаване на постсоциалистически елити разора почвата за поникването на хибридни модели на демокрации, чиито последствия виждаме днес.

България в Европа и Европа срещу света – къде сме в технологичното състезание?
Аналогово спокойствие в съвършено обезпечен дигитален свят звучи като светло бъдеще. За да го постигнем, трябва да си го представим, а после и да участваме активно в изграждането му. Засега ЕС и България изостават от технологичната надпревара. Каква е общата картина – от Александър Нуцов.
Обединена Европа и европейската мечтаТоестАлександър Нуцов
Обединена Европа и европейската мечта

В този смисъл упадъкът в европейската външна политика започна, когато ЕС се опита да се върне към имперското си мислене чрез трансформирането на нагласите и културните ценности на своите държави членки. Така се сбъдна историческото пророчество на диалектиката на Просвещението: че Европа ще се завърне към имперското си минало, както и САЩ, които направиха няколко несполучливи геополитически експеримента в Близкия изток. 

Ето защо първата стъпка към истинското обединение на Европа и Запада е именно отказът от стратегическата култура на империализма, която цели овладяването на цялото земно кълбо под флага на една идеология. Ако се замислим, така се провали и СССР и можем само да се надяваме, че Европейският съюз няма да допусне същата грешка.

И тук идват следващите крачки за изпълнението на тази цел, при които трябва да сме наясно с една много важна разлика – отказът от империалистката стратегическа култура не следва да се търси нито в крайнолевите, нито в крайнодесните идеологии на нашето време. Да се откажеш от такова мислене означава две неща. Първо, да спреш да изнасяш ценности, опитвайки се да ги наложиш по цялото земно кълбо, претендирайки, че те са универсални, общовалидни и носители на мир. Второ, да загърбиш расизма, ксенофобията, антисемитизма и всички пороци на миналото, които спиралата на историята е осъдила на трибунала в Нюрнберг, в съдебния процес срещу Айхман и на международните трибунали срещу геноцида в Руанда и Сребреница. 

Пътят към постигането на тази цел минава през секуларизацията на европейската политика и проектирането на сила в рамките на европейската отбранителна идентичност единствено в Западното полукълбо. Уви, това неизбежно ще наложи както свиването на средната класа в Европа, така и отказ от по-нататъшно разширяване на ЕС. За жалост, на много от тези цели днес се гледа с огромна доза недоверие и дори скептицизъм, тъй като идеологическият порив, за който говорят Хоркхаймер и Адорно, очевидно отново е обзел Стария континент.

И все пак къде е днес Европа?

Нашият съюз днес е вплетен в рамките на геополитическата надпревара, като европейските елити дори не подозират, че невъзможността ни да се защитаваме е добра новина както за Китай и Русия, така и за САЩ. За да създадем отбранителни способности, са необходими преди всичко много пари, но най-вече формиране на нова стратегическа култура, която да положи основите на федерализацията на Европа. Важно е да подчертаем, че за да се предпази от „изкушенията“, за които говорят Хоркхаймер и Адорно – неоколониализъм, културна хегемония, ценностен универсализъм, реакционен традиционализъм, – ЕС следва преди всичко да се федерализира икономически и военно, а не политически. 

Политическата интеграция на Европа може да се окаже опасен ход, който да я върне обратно в 30-те години на XX век, а и обективно погледнато, толкова голям политически консенсус все още няма. Статистиките сочат такъв консенсус единствено в Прогресивния алианс на социалистите и демократите, а в Европейската народна партия нагласите са по-скоро разделени. По сходен начин стоят нещата в Германия и Италия, където общественото мнение е разделено, а значителен дял от французите все още не смятат, че Европа е готова да се федерализира. 

(Не)демократичният световен ред
Защо е необходимо да се борим за оцеляването на демокрацията? Надали само защото нищо по-добро не е измислено. От Искрен Иванов.
Обединена Европа и европейската мечтаТоестИскрен Иванов
Обединена Европа и европейската мечта

Официалните данни на Евробарометър не показват широк обществен консенсус за по-дълбока политическа интеграция. В ключови държави, като Германия, Италия и Франция, подкрепата за ЕС остава колеблива, доверието към институциите е около или под 50%, а мнозинството от гражданите не смята, че ЕС се движи в правилната посока. Дори при висока подкрепа за отделни общи политики – като европейска отбрана – липсва обществена основа за скок към федерална политическа структура.

Следователно, за да се избегне изкушението на идеологическите крайности, е добре Европа преди всичко да се погрижи за своята икономическа и военна интеграция, което ще примири различията между крайнолеви и крайнодесни и ще положи основите на по-силен и обединен политически съюз. 

От друга страна, бързата федерализация на ЕС, паралелно с трупането на военна мощ, ще го превърне в нова империя, която ще може да използва силите си не само да се отбранява, но и да напада. Поляризацията в европейските общества ще засили разделението в страните членки и ще създаде политически климат, подобен на американския, заменяйки терористичните атентати с постоянни сблъсъци между европейците, които могат да доведат до появата на мощни дезинтеграционни процеси в Общността. И така, в един момент „лекарството“ може да се окаже дори по-лошо от болестта.

И накрая, нека си дадем сметка, че за държави като нашата оцеляването на ЕС е въпрос както на националната ни сигурност, така и на съхраняването на собствения ни икономически суверенитет. Европа е оцеляла, защото се е реформирала и само една мащабна реформа в учредителните договори на Общността може да открехне вратата към преодоляването на демократичния дефицит в европейските институции и възвръщането на доверието в тях. 

За Китай, Европа и червените линии
В няколко поредни материала ще се опитаме да разгледаме малко по-задълбочено Китай и отношенията му с Европа, САЩ и Русия. Първият от тях е посветен на концепцията за китайската външна политика и как тя изглежда в отношенията на Китай с Европа. От Искрен Иванов.
Обединена Европа и европейската мечтаТоестИскрен Иванов
Обединена Европа и европейската мечта

България и Източна Европа като цяло могат да се превърнат в мотор на тези процеси, тъй като според някои изследвания Централна Европа и Източният фланг се справят много по-добре с вътрешната си сигурност, отколкото западните ни партньори, и това е отразено и в препоръките на американското правителство. Източноевропейците, които в продължение на години бяха обект на предразсъдъци и дискриминация от страна на останалите държави в Общността, днес може би държат ключа за възстановяването на Европа преди всичко като зона на сигурност и стабилност. 

Уви, винаги стои и предупреждението на Хоркхаймер и Адорно, че свръхфокусирането на Източна Европа повече върху историческата справедливост, отколкото върху свободата, може да доведе до тоталитаризъм – единствената по-страшна алтернатива на империализма. И никоя източноевропейска държава не е застрахована от такова развитие, ако бъде оставена в европейската периферия като „мост“ между Изтока и Запада.

AMD Intros Kintex UltraScale+ Gen 2 FPGAs

Post Syndicated from Ryan Smith original https://www.servethehome.com/amd-intros-kintex-ultrascale-gen-2-fpgas/

For this year’s Integrated Systems Europe trade show, AMD is unveiling an upgraded lineup of its Kintex mid-range FPGAs. Dubbed the Kintex UltraScale+ Gen 2 family, this latest iteration of the Kintex family is designed to modernize the FPGAs, bringing support for newer and faster memory, I/O, and cryptography technologies. As with the existing (Gen […]

The post AMD Intros Kintex UltraScale+ Gen 2 FPGAs appeared first on ServeTheHome.

Build an AI-powered course recommender using Amazon Bedrock and AWS End User Messaging

Post Syndicated from Ruchikka Chaudhary original https://aws.amazon.com/blogs/messaging-and-targeting/build-an-ai-powered-course-recommender-using-amazon-bedrock-and-aws-end-user-messaging/

Educational technology (EdTech) providers face the challenge of maintaining seamless, personalized communication and presenting the right recommendations to their diverse stakeholders. This post explores how combining Amazon Web Services (AWS) End User Messaging and WhatsApp Business API with the advanced AI capabilities of Amazon Bedrock can transform educational engagement.

In this post, we explore use cases that are reshaping the EdTech industry. We discover how application automation can streamline admissions and enrollment processes, making them more efficient and user-friendly. We demonstrate how instant student engagement can be achieved through AI-powered, personalized interactions that keep learners motivated and connected. We showcase how real-time course feedback mechanisms can help educators adapt and improve their teaching methods. We also examine how student support can be automated using intelligent assistants that provide continuous, all-day assistance while maintaining a personal touch.

We show you how to build an AI-powered course recommendation system. We explain how to set up WhatsApp Business API integration with Amazon Bedrock, implement smart search capabilities for course matching, and create a scalable serverless architecture. You’ll learn how to build meaningful analytics dashboards to track engagement and learn best practices for handling errors and maintaining system reliability. Whether you’re an EdTech professional or a cloud architect, this guide gives you practical insights into combining conversational AI with educational services.

Use cases

  • An AI-powered personalized learning pathway generator that automatically recommends customized content based on individual student performance metrics and learning requirements
  • Course improvement suggestions and real-time course feedback
  • A smart communication orchestrator that delivers role-specific, automated notifications and updates across multiple channels to enhance student and parent engagement
  • An early warning system using predictive analytics to identify at-risk students through real-time monitoring of engagement metrics and performance indicators
  • Student support automation with always available AI assistant support, FAQ handling, escalation management, and multilingual support

Prerequisites

  • An AWS account
  • AWS End User Messaging set up with WhatsApp channel enabled
  • A pre-existing WhatsApp Business account
  • Amazon Bedrock setup must be completed with preferred model
  • Amazon Quick Sight for the AWS Region must be enabled

Solution overview

With this solution, users can discover and order educational courses through WhatsApp conversations. Instead of navigating complex websites, the user can send a WhatsApp message saying, “I want to learn Python programming.” They’ll receive personalized course recommendations instantly. The architecture processes WhatsApp messages through AWS End User Messaging, uses Amazon Bedrock for AI-powered conversations, performs semantic search with Amazon OpenSearch Serverless, and captures analytics for business insights. (For step-by-step implementation and rollback guidelines, see the sample course recommendation system.) The following architectural diagram illustrates a modern AI-powered course recommendation system that uses multiple AWS services.

Figure 1: AI-powered course recommendation system

Message processing

When users send WhatsApp messages, AWS End User Messaging captures them and publishes events to an Amazon Simple Notification Service (Amazon SNS) topic. This creates a decoupled architecture where multiple services can process the same message events independently. AWS Lambda functions subscribe to these events, facilitating reliable message processing during high-traffic periods. The decoupled design provides several advantages:

  • If one component fails, others continue operating
  • You can add new message processors without affecting existing ones
  • The system automatically scales based on message volume without manual intervention

AI conversation engine

Amazon Bedrock with Claude 3 Haiku powers natural language understanding. It is configured specifically for WhatsApp with instructions for short paragraphs, relevant emoji, and mobile-optimized responses.

AI agents

The agent maintains conversation context and handles structured actions such as course search, detail retrieval, and booking through defined functions. The following workflow is the agent action flow and sample code:

Agent flow

  1. Greets user → Understands intent → Searches courses → Provides details → Facilitates booking
  2. Maintains context throughout the conversation
  3. Can switch between actions based on user responses
  4. Handles complex queries by combining multiple actions

Sample code

The following is sample code to create a Bedrock agent using AWS CDK:

    agent = bedrock.CfnAgent(foundation_model="anthropic.claude-3-haiku-20240307-v1:0",
     instruction="""
     Format for WhatsApp: short paragraphs,
        focus on technical courses only
       """,
      action_groups=[# Functions for search, details, booking]
)

Semantic search

Traditional keyword search can miss the user’s intent. The application uses Amazon Titan Embeddings in Amazon Bedrock to convert courses and queries into vectors, enabling semantic understanding. When users ask for “cloud computing courses,” the system can understand related terms such as “AWS” and “serverless” without exact matches. Amazon OpenSearch Serverless handles vector similarity matching combined with traditional filters for course price, level, and duration.

Analytics pipeline

Every WhatsApp message interaction generates business intelligence. Messages are stored in Amazon Simple Storage Service (Amazon S3) with date partitioning, catalogued through AWS Glue, and made queryable using Amazon Athena. Teams can analyze user behavior, popular topics, and conversion rates through Quick Sight dashboards. The following dashboard shows example widgets displaying pie-chart breakdown of message delivery status and count of messages per day.

Figure 2: Amazon Quick Sight dashboard

As shown in the following dashboard, Amazon Q in QuickSight enables you to explore and analyze your data using conversational AI capabilities.

Figure 3: Amazon Quick Sight dashboard showing chat window

Error handling and resilience

Such highly scalable and distributed solutions require robust error handling. The application has exponential backoff and retries for API calls, meaning the system can gracefully handle rate limits and temporary service unavailability.

The following is sample code for error handling and resilience:

python
def retry_with_backoff(func, max_retries=5):
retries = 0
backoff = 1
while retries < max_retries:
try:
return func()
except ThrottlingException:
sleep_time = backoff + random.uniform(0, 1)
time.sleep(sleep_time)
backoff = min(backoff * 2, 32)
retries += 1
raise Exception("Max retries exceeded")

Business impact

With the global EdTech market expected to reach $165 billion by 2026, educators and institutions are seeking solutions to prevent student dropouts, improve learning outcomes, and maintain their competitive advantage. Poor personalization can lead to decreased student engagement, lower course completion rates, and ultimately revenue loss.

Implementing AI-driven personalization and communication systems means institutions can significantly improve student retention rates, boost learning outcomes, and create a more engaging educational experience, which directly impacts their bottom line and reputation in an increasingly competitive educational landscape. This solution could transform educational delivery through intelligent personalization and operational excellence. A serverless architecture can help educational institutions focus on content quality rather than infrastructure management while potentially maintaining rapid response times for course searches. The system’s analytics capabilities could offer insights into student behavior and course preferences, helping shape future curriculum development.

With mobile optimization, institutions can better serve the growing population of digital-first learners. The combination of automated scaling and pay-per-use pricing could create opportunities for cost optimization, and real-time dashboards can be used to facilitate data-informed decision-making. Such improvements in user experience and operational efficiency could lead to enhanced student engagement and institutional growth in the evolving education environment.

Sample conversation

The following video shows how a user can interact with the generative AI-powered course recommendation system and receive course recommendations.

Future enhancements

We’re expanding to more messaging platforms, adding voice integration through Amazon Connect, and implementing predictive analytics for personalized recommendations. The serverless architecture makes these additions straightforward without infrastructure changes. Future scenarios could involve:

  • Educator and student support – This solution can be enhanced for student and educator experiences. For educators, it can automate administrative tasks. For students, it can create personalized engagement campaigns, a communication approach that could be significantly more effective than traditional methods.
  • Digital admission process flow – The solution integrates AWS Bedrock AI with WhatsApp Business API to streamline digital admissions. It can enable instant document verification, guide secure payments, and provide automated updates, all within the AWS End User Messaging WhatsApp channel. This AI-powered system could transform the complex admission process into an efficient, chat-based experience, benefiting both institutions and applicants.
  • Parental support and study material management – The system could intelligently distribute learning resources based on student needs, send automated schedule updates, and provide personalized progress reports to parents through WhatsApp. Parents could receive AI-curated study materials and real-time updates about their child’s academic performance, homework assignments, and upcoming assessments through familiar chat interactions. This integration could transform traditional parent-teacher communication into an efficient, automated system while providing timely access to relevant educational resources.

Conclusion

The WhatsApp course recommender agent demonstrates how modern AWS services can create sophisticated, AI-powered conversational experiences that scale automatically and provide rich business insights. The serverless architecture provides cost-effectiveness while maintaining enterprise-grade reliability. Key architectural principles that make this solution successful include event-driven design for scalability, AI integration for natural interactions, semantic search for superior user experience, customizable analytics for business intelligence, and infrastructure as code (IaC) for reliable deployments.

For organizations considering similar implementations, we recommend focusing on user experience optimization, robust error handling, comprehensive monitoring, and gradual feature rollout. The conversational AI environment is rapidly evolving, and solutions that prioritize user experience while maintaining technical excellence can drive the most business value. This implementation can serve as a reference architecture for building production-ready conversational AI systems on AWS, demonstrating patterns that can apply across industries and use cases.

About the authors

AWS IAM Identity Center now supports multi-Region replication for AWS account access and application use

Post Syndicated from Channy Yun (윤석찬) original https://aws.amazon.com/blogs/aws/aws-iam-identity-center-now-supports-multi-region-replication-for-aws-account-access-and-application-use/

Today, we’re announcing the general availability of AWS IAM Identity Center multi-Region support to enable AWS account access and managed application use in additional AWS Regions.

With this feature, you can replicate your workforce identities, permission sets, and other metadata in your organization instance of IAM Identity Center connected to an external identity provider (IdP), such as Microsoft Entra ID and Okta, from its current primary Region to additional Regions for improved resiliency of AWS account access.

You can also deploy AWS managed applications in your preferred Regions, close to application users and datasets for improved user experience or to meet data residency requirements. Your applications deployed in additional Regions access replicated workforce identities locally for optimal performance and reliability.

When you replicate your workforce identities to an additional Region, your workforce gets an active AWS access portal endpoint in that Region. This means that in the unlikely event of an IAM Identity Center service disruption in its primary Region, your workforce can still access their AWS accounts through the AWS access portal in an additional Region using already provisioned permissions. You can continue to manage IAM Identity Center configurations from the primary Region, maintaining centralized control.

Enable IAM Identity Center in multiple Regions
To get started, you should confirm that the AWS managed applications you’re currently using support customer managed AWS Key Management Service (AWS KMS) key enabled in AWS Identity Center. When we introduced this feature in October 2025, Seb recommended using multi-Region AWS KMS keys unless your company policies restrict you to single-Region keys. Multi-Region keys provide consistent key material across Regions while maintaining independent key infrastructure in each Region.

Before replicating IAM Identity Center to an additional Region, you must first replicate the customer managed AWS KMS key to that Region and configure the replica key with the permissions required for IAM Identity Center operations. For instructions on creating multi-Region replica keys, refer to Create multi-Region replica keys in the AWS KMS Developer Guide.

Go to the IAM Identity Center console in the primary Region, for example, US East (N. Virginia), choose Settings in the left-navigation pane, and select the Management tab. Confirm that your configured encryption key is a multi-Region customer managed AWS KMS key. To add more Regions, choose Add Region.

You can choose additional Regions to replicate the IAM Identity Center in a list of the available Regions. When choosing an additional Region, consider your intended use cases, for example, data compliance or user experience.

If you want to run AWS managed applications that access datasets limited to a specific Region for compliance reasons, choose the Region where the datasets reside. If you plan to use the additional Region to deploy AWS applications, verify that the required applications support your chosen Region and deployment in additional Regions.

Choose Add Region. This starts the initial replication whose duration depends on the size of your Identity Center instance.

After the replication is completed, your users can access their AWS accounts and applications in this new Region. When you choose View ACS URLs, you can view SAML information, such as an Assertion Consumer Service (ACS) URL, about the primary and additional Regions.

How your workforce can use an additional Region
AWS Identity Center supports SAML single sign-on with external IdPs, such as Microsoft Entra ID and Okta. Upon authentication in the IdP, the user is redirected to the AWS access portal. To enable the user to be redirected to the AWS access portal in the newly added Region, you need to add the additional Region’s ACS URL to the IdP configuration.

The following screenshots show you how to do this in the Okta admin console:

Then, you can create a bookmark application in your identity provider for users to discover the additional Region. This bookmark app functions like a browser bookmark and contains only the URL to the AWS access portal in the additional Region.

You can also deploy AWS managed applications in additional Regions using your existing deployment workflows. Your users can access applications or accounts using the existing access methods, such as the AWS access portal, an application link, or through the AWS Command Line Interface (AWS CLI).

To learn more about which AWS managed applications support deployment in additional Regions, visit the IAM Identity Center User Guide.

Things to know
Here are key considerations to know about this feature:

  • Consideration – To take advantage of this feature at launch, you must be using an organization instance of IAM Identity Center connected to an external IdP. Also, the primary and additional Regions must be enabled by default in an AWS account. Account instances of IAM Identity Center, and the other two identity sources (Microsoft Active Directory and IAM Identity Center directory) are presently not supported.
  • Operation – The primary Region remains the central place for managing workforce identities, account access permissions, external IdP, and other configurations. You can use the IAM Identity Center console in additional Regions with a limited feature set. Most operations are read-only, except for application management and user session revocation.
  • Monitoring – All workforce actions are emitted in AWS CloudTrail in the Region where the action was performed. This feature enhances account access continuity. You can set up break-glass access for privileged users to access AWS if the external IdP has a service disruption.

Now available
AWS IAM Identity Center multi-Region support is now available in the 17 enabled-by-default commercial AWS Regions. For Regional availability and a future roadmap, visit the AWS Capabilities by Region. You can use this feature at no additional cost. Standard AWS KMS charges apply for storing and using customer managed keys.

Give it a try in the AWS Identity Center console. To learn more, visit the IAM Identity Center User Guide and send feedback to AWS re:Post for Identity Center or through your usual AWS Support contacts.

— Channy

Use Amazon MSK Connect and Iceberg Kafka Connect to build a real-time data lake

Post Syndicated from Xiao Huang original https://aws.amazon.com/blogs/big-data/use-amazon-msk-connect-and-iceberg-kafka-connect-to-build-a-real-time-data-lake/

As analytical workloads increasingly demand real-time insights, organizations need business data to enter the data lake immediately after generation. While various methods exist for real-time CDC data ingestion (such as AWS Glue and Amazon EMR Serverless), Amazon MSK Connect with Iceberg Kafka Connect provides a fully managed, streamlined approach that reduces operational complexity and enables continuous data synchronization.

In this post, we demonstrate how to use Iceberg Kafka Connect with Amazon Managed Streaming for Apache Kafka (Amazon MSK) Connect to accelerate real-time data ingestion into data lakes, simplifying the synchronization process from transactional databases to Apache Iceberg tables.

Solution overview

In this post, we show you how to implement capturing transaction log data from Amazon Relational Database Service (Amazon RDS) for MySQL and writing it to Amazon Simple Storage Service (Amazon S3) in Iceberg table format using append mode, covering both single-table and multi-table synchronization, as shown in the following figure.

Downstream consumers then process these change records to reconstruct the data state before writing to Iceberg tables.

In this solution, you use the Iceberg Kafka Sink Connector to implement the business on the sink side. The Iceberg Kafka Sink Connector has the following features:

  • Supports exactly-once delivery
  • Support multi-table synchronization
  • Support schema changes
  • Field name mapping through Iceberg’s column mapping feature

Prerequisites

Before beginning the deployment, ensure you have the following components in place:

Amazon RDS for MySQL: This solution assumes you already have an Amazon RDS for MySQL database instance running with the data you want to synchronize to your Iceberg data lake. Ensure that binary logging is enabled on your RDS instance to support Change Data Capture (CDC) operations.

Amazon MSK Cluster: You need an Amazon MSK cluster provisioned in your target AWS Region. This cluster will serve as the streaming platform between your MySQL database and the Iceberg data lake. Ensure the cluster is properly configured with appropriate security groups and network access.

Amazon S3 Bucket: Ensure you have an Amazon S3 bucket ready to host the custom Kafka Connect plugins. This bucket serves as the storage location from which AWS MSK Connect retrieves and installs your plugins. The bucket must exist in your target AWS Region, and you must have appropriate permissions to upload objects to it.

Custom Kafka Connect Plugins: To enable real-time data synchronization with MSK Connect, you need to create two custom plugins. The first plugin uses the Debezium MySQL Connector to read transactional logs and produce Change Data Capture (CDC) events. The second plugin uses Iceberg Kafka Connect to synchronize data from Amazon MSK to Apache Iceberg tables.

Build Environment: To build the Iceberg Kafka Connect plugin, you need a build environment with Java and Gradle installed. You can either launch an Amazon EC2 instance (recommended: Amazon Linux 2023 or Ubuntu) or use your local machine if it meets the requirements. Ensure you have sufficient disk space (at least 20GB) and network connectivity to clone the repository and download dependencies.

Build Iceberg Kafka Connect from open source

The connector ZIP archive is created as part of the Iceberg build. You can run the build using the following code:

git clone https://github.com/apache/iceberg.git
cd iceberg/
./gradlew -x test -x integrationTest clean build
The ZIP archive will be saved in ./kafka-connect/kafka-connect-runtime/build/distributions.

Create custom plugins

The next step is to create custom plugins to read and synchronize the data.

  1. Upload the custom plugin ZIP file you compiled in the previous step to your designated Amazon S3 bucket.
  2. Go to the AWS Management Console and navigate to Amazon MSK and choose Connect in the navigation pane.
  3. Choose Custom plugins, then select the plugin file you uploaded to S3 by browsing or entering its S3 URI.
  4. Specify a unique, descriptive name for your custom plugin (such as my-connector-v1).
  5. Choose Create custom plugin.

Configure MSK Connect

With the plugins installed, you’re ready to configure MSK Connect.

Configure data source access

Start by configuring data source access.

  1. To create a worker configuration, choose Worker configurations in the MSK Connect console.
  2. Choose Create worker configuration and copy and paste the following configuration. 
    key.converter.schemas.enable=false
value.converter.schemas.enable=false
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
# Enable topic creation by the worker
topic.creation.enable=true
# Default topic creation settings for debezium connector
topic.creation.default.replication.factor=3
topic.creation.default.partitions=1
topic.creation.default.cleanup.policy=delete

  3. In the Amazon MSK console, choose Connectors under Amazon MSK Connect and choose Create connector.
  4. In the setup wizard, select the Debezium MySQL Connector plugin created in the previous step, enter the connector name and select the MSK cluster of the synchronization target. Copy and paste the following content in the configuration: 
    
connector.class=io.debezium.connector.mysql.MySqlConnector
tasks.max=1
include.schema.changes=false
database.server.id=100000
database.server.name=
database.port=3306
database.hostname=
database.password=
database.user=

topic.creation.default.partitions=1
topic.creation.default.replication.factor=3

topic.prefix=mysqlserver
database.include.list=

## route
transforms=Reroute
transforms.Reroute.type=io.debezium.transforms.ByLogicalTableRouter
transforms.Reroute.topic.regex=(.*)(.*)
transforms.Reroute.topic.replacement=$1all_records

# schema.history
schema.history.internal.kafka.topic
schema.history.internal.kafka.bootstrap.servers=
# IAM/SASL
schema.history.internal.consumer.sasl.mechanism=AWS_MSK_IAM
schema.history.internal.consumer.sasl.client.callback.handler.class=software.amazon.msk.auth.iam.IAMClientCallbackHandler
schema.history.internal.consumer.security.protocol=SASL_SSL
schema.history.internal.consumer.sasl.jaas.config=software.amazon.msk.auth.iam.IAMLoginModule required;
schema.history.internal.producer.security.protocol=SASL_SSL
schema.history.internal.producer.sasl.mechanism=AWS_MSK_IAM
schema.history.internal.producer.sasl.client.callback.handler.class=software.amazon.msk.auth.iam.IAMClientCallbackHandler
schema.history.internal.producer.sasl.jaas.config=software.amazon.msk.auth.iam.IAMLoginModule required;

    Note that in the configuration, Route is used to write multiple records to the same topic. In the parameter transforms.Reroute.topic.regex, the regular expression is configured to filter the table names that need to be written to the same topic. In the following example, the data containing <tablename-prefix> in the table name is written to the same topic.

    ## route
transforms=Reroute
transforms.Reroute.type=io.debezium.transforms.ByLogicalTableRouter
transforms.Reroute.topic.regex=(.*)(.*)
transforms.Reroute.topic.replacement=$1all_records

    For example, after transforms.Reroute.topic.replacement is specified as $1all_records, the topic name created in the MSK is < database.server.name>.all_records.

  5. After you choose Create, MSK Connect creates a synchronization task for you.

Data synchronization (single table mode)

Now, you can create a real-time synchronization task for the Iceberg table. Start by creating a real-time synchronization job for a single table.

  1. In the Amazon MSK console, choose Connectors under MSK Connect
  2. Choose Create connector.
  3. On the next page, select the previously created Iceberg Kafka Connect plugin
  4. Enter the connector name and select the MSK cluster of the synchronization target.
  5. Paste the following code in the configuration. 
    
connector.class=org.apache.iceberg.connect.IcebergSinkConnector
tasks.max=1
topics=
iceberg.tables=
iceberg.catalog.catalog-impl=org.apache.iceberg.aws.glue.GlueCatalog
iceberg.catalog.warehouse=
iceberg.catalog.io-impl=org.apache.iceberg.aws.s3.S3FileIO
iceberg.catalog.client.region=
iceberg.tables.auto-create-enabled=true
iceberg.tables.evolve-schema-enabled=true
iceberg.control.commit.interval-ms=120000
transforms=debezium
transforms.debezium.type=org.apache.iceberg.connect.transforms.DebeziumTransform
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
key.converter.schemas.enable=false
iceberg.control.topic=control-iceberg

    For Iceberg Connector, it will create a topic named control-iceberg by default to record offset. Select the previously created worker configuration that includes topic.creation.enable = true. If you use the default worker configuration and auto-topic creation isn’t enabled at the MSK broker level, the connector will not be able to automatically create topics.

    You can also specify this topic name by setting the parameter iceberg.control.topic = <offset-topic>. If you want to use a custom topic, you can use the following code.

    $KAFKA_HOME/bin/kafka-topics.sh --bootstrap-server $MYBROKERS --create --topic <my-iceberg-offset-topic> --partitions 3 --replication-factor 2 --config cleanup.policy=compact

  6. Query the synchronized data results through Amazon Athena. From the table synchronized to Athena, you can see that, in addition to the source table field, an additional _cdc field has been added to store the metadata content of the CDC.

Compaction

Compaction is an essential maintenance operation for Iceberg tables. Although frequent ingestion of small files can negatively impact query performance, regular compaction mitigates this issue by consolidating small files, minimizing metadata overhead, and substantially improving query efficiency. To maintain optimal table performance, you should implement dedicated compaction workflows. AWS Glue offers an excellent solution for this purpose, providing automated compaction capabilities that intelligently merge small files and restructure table layouts for enhanced query performance.

Schema Evolution Demonstration

To demonstrate the schema evolution capabilities of this solution, we conducted a test to show how field changes at the source database are automatically synchronized to the Iceberg tables through MSK Connect and Iceberg Kafka Connect.

Initial Setup:

First, we created an RDS MySQL database with a customer information table (tb_customer_info) containing the following schema:

+----------------+--------------+------+-----+-------------------+-----------------------------------------------+
| Field          | Type         | Null | Key | Default           | Extra                                         |
+----------------+--------------+------+-----+-------------------+-----------------------------------------------+
| id             | int unsigned | NO   | PRI | NULL              | auto_increment                                |
| user_name      | varchar(64)  | YES  |     | NULL              |                                               |
| country        | varchar(64)  | YES  |     | NULL              |                                               |
| province       | mediumtext   | NO   |     | NULL              |                                               |
| city           | int          | NO   |     | NULL              |                                               |
| street_address | varchar(20)  | NO   |     | NULL              |                                               |
| street_name    | varchar(20)  | NO   |     | NULL              |                                               |
| created_at     | timestamp    | NO   |     | CURRENT_TIMESTAMP | DEFAULT_GENERATED                             |
| updated_at     | timestamp    | YES  |     | CURRENT_TIMESTAMP | DEFAULT_GENERATED on update CURRENT_TIMESTAMP |
+----------------+--------------+------+-----+-------------------+-----------------------------------------------+

We then configured MSK Connect using the Debezium MySQL Connector to capture changes from this table and stream them to Amazon MSK in real time. Following that, we set up Iceberg Kafka Connect to consume the data from MSK and write it to Iceberg tables.

Schema Modification Test:

To test the schema evolution capability, we added a new field named phone to the source table:

ALTER TABLE tb_customer_info ADD COLUMN phone VARCHAR(20) NULL;

We then inserted a new record with the phone field populated:

INSERT INTO tb_customer_info (user_name,country,province,city,street_address,street_name,phone) values ('user_demo','China','Guangdong',755,'Street1 No.369','Street1','13099990001');

Results:

When we queried the Iceberg table in Amazon Athena, we observed that the phone field had been automatically added as the last column, and the new record was successfully synchronized with all field values intact. This demonstrates that Iceberg Kafka Connect’s self-adaptive schema capability seamlessly handles DDL changes at the source, eliminating the need for manual schema updates in the data lake.

Data synchronization (multi-table mode)

It’s common that data admins want to use a single connector for moving data in multiple tables. For example, you can use the CDC collection tool to write data from multiple tables to a topic and then write data from one topic to multiple Iceberg tables through the consumer side. In Configure data source access, you configured a MySQL synchronization Connector to synchronize tables with specified rules to a topic using Route. Now let’s review how to distribute data from this topic to multiple Iceberg tables.

  1. When using Iceberg Kafka Connect to synchronize multiple tables to Iceberg tables using AWS Glue Data Catalog, you must pre-create a database in the Data Catalog before starting the synchronization process. The database name in AWS Glue must exactly match the source database name, because the Iceberg Kafka Connect connector automatically uses the source database name as the target database name during multi-table synchronization. This naming consistency is required because the connector doesn’t provide an option to map source database names to different target database names in multi-table scenarios.
  2. If you want to use your custom topic name, you can create a new topic to store the MSK Connect record offset, see Data synchronization (single table mode).
  3. In the Amazon MSK console, create another connector using the following configuration. 
    connector.class= org.apache.iceberg.connect.IcebergSinkConnector
tasks.max=2
topics=
iceberg.catalog.catalog-impl=org.apache.iceberg.aws.glue.GlueCatalog
iceberg.catalog.warehouse=
iceberg.catalog.io-impl=org.apache.iceberg.aws.s3.S3FileIO
iceberg.catalog.client.region=
iceberg.tables.auto-create-enabled=true
iceberg.tables.evolve-schema-enabled=true
iceberg.control.commit.interval-ms=120000
transforms=debezium
transforms.debezium.type=org.apache.iceberg.connect.transforms.DebeziumTransform
iceberg.tables.route-field=_cdc.source
iceberg.tables.dynamic-enabled=true
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
key.converter.schemas.enable=false
iceberg.control.topic=control-iceberg

    In this configuration, two parameters have been added:

    • iceberg.tables.route-field: Specifies the routing field that distinguishes between different tables, specified as cdc.source for CDC data parsed by Debezium
    • iceberg.tables.dynamic-enabled: If the iceberg.tables parameter isn’t set, it must be specified as true here
  4. After completion, MSK Connect will creates a sink connector for you.
  5. After the process is complete, you can view the newly created table through Athena.

Other tips

In this section, we share some more things that you can use to customize your deployment to fit your use case.

  • Specified table synchronizationIn the Data synchronization (multi-table mode) section, you specify iceberg.tables.route-field = _cdc.Source and iceberg.tables.dynamic-enabled=true, these two parameter settings can write multiple tables stored in the Iceberg table. If you want to synchronize only the specified tables, you can specify the table name you want to synchronize by setting iceberg.tables.dynamic-enabled = false and then setting the iceberg.tables parameter. For example, 
    iceberg.tables.dynamic-enabled = false
iceberg.tables = default.tablename1,default.tablename2
 
iceberg.table.default.tablename1.route-regex = tablename1
iceberg.table.default.tablename2.route-regex = tablename2

  • Performance Testing Results
    We conducted a performance test using sysbench to evaluate the data synchronization capabilities of this solution. The test simulated a high-volume write scenario to demonstrate the system’s throughput and scalability.Test Configuration:

    1. Database setup: Created 25 tables in the MySQL database using sysbench
    2. Data loading: Wrote 20 million records to each table (500 million total records)
    3. Real-time streaming: Configured MSK Connect to stream data from MySQL to Amazon MSK in real time during the write process
    4. Kafka Connect configuration:
      • Started Kafka Iceberg Connect
      • Minimum workers: 1
      • Maximum workers: 8
      • Allocated two MCUs per worker

    Performance Results:

    In our test using the configuration above, each MCU achieved peak writing performance of approximately 10,000 records per second, as shown in the following figure. This demonstrates the solution’s ability to handle high-throughput data synchronization workloads effectively.

Clean up

To clean up your resources, complete the following steps:

  1. Delete MSK Connect connectors: Remove both the Debezium MySQL Connector and Iceberg Kafka Connect connector created for this solution.
  2. Delete the Amazon MSK cluster: If you created a new MSK cluster specifically for this demonstration, delete it to stop incurring charges.
  3. Delete the S3 buckets: Remove the S3 buckets used to store the custom Kafka Connect plugins and Iceberg table data. Ensure you have backed up any data you need before deletion.
  4. Delete the EC2 instance: If you launched an EC2 instance to build the Iceberg Kafka Connect plugin, terminate it.
  5. Delete the RDS MySQL instance (optional): If you created a new RDS instance specifically for this demonstration, delete it. If you’re using an existing production database, skip this step.
  6. Remove IAM roles and policies (if created): Delete any IAM roles and policies that were created specifically for this solution to maintain security best practices.

Conclusion

In this post, we presented a solution to achieve real-time, efficient data synchronization from transactional databases to data lakes using Amazon MSK Connect and Iceberg Kafka Connect. This solution provides a low-cost and efficient data synchronization paradigm for enterprise-level big data analysis. Whether you’re working with ecommerce transactions, financial transactions, or IoT device logs, this solution can help you achieve quick access to a data lake, enabling analytical businesses to quickly obtain the latest business data. We encourage you to try this solution in your own environment and share your experiences in the comments section. For more information, visit Amazon MSK Connect.

About the author

Huang Xiao

Huang Xiao

Huang is a Senior Specialist Solution Architect with Analytics at AWS. He focuses on big data solution architecture design, with years of experience in development and architectural design within the big data field.

Optimizing Flink’s join operations on Amazon EMR with Alluxio

Post Syndicated from Qingyuan Tang original https://aws.amazon.com/blogs/big-data/optimizing-flinks-join-operations-on-amazon-emr-with-alluxio/

When you’re working with data analysis, you often face the challenge of effectively correlating real-time data with historical data to gain actionable insights. This becomes particularly critical when you’re dealing with scenarios like e-commerce order processing, where your real-time decisions can significantly impact business outcomes. The complexity arises when you need to combine streaming data with static reference information to create a comprehensive analytical framework that supports both your immediate operational needs and strategic planning

To tackle this challenge, you can employ stream processing technologies that handle continuous data flows while seamlessly integrating live data streams with static dimension tables. These solutions enable you to perform detailed analysis and aggregation of data, giving you a comprehensive view that combines the immediacy of real-time data with the depth of historical context. Apache Flink has emerged as a leading stream computing platform that offers robust capabilities for joining real-time and offline data sources through its extensive connector ecosystem and SQL API.

In this post, we show you how to implement real-time data correlation using Apache Flink to join streaming order data with historical customer and product information, enabling you to make informed decisions based on comprehensive, up-to-date analytics.

We also introduce an optimized solution to automatically load Hive dimension table data into Alluxio Universal Flash Storage (UFS) through the Alluxio cache layer. This enables Flink to perform temporal joins on changing data, accurately reflecting the content of a table at specific points in time.

Solution architecture

When it comes to joining Flink SQL tables with stream tables, the lookup join is a go-to method. This approach is particularly effective when you need to correlate streaming data with static or slowly changing data. In Flink, you can use connectors like the Flink Hive SQL connector or the FileSystem connector to archive the scenario.

The following architecture shows general approach which we describe ahead：

Here’s how we do this:

  1. We use offline data to construct a Flink table. This data could be from an offline Hive database table or from files stored in a system like Amazon S3. Concurrently, we can create a stream table from the data flowing in through a Kafka message stream
  2. Use a batch cluster for offline data processing. In this example, we use an Amazon EMR cluster which creates a fact table in it. It also provides a Detail Wide Data (DWD) table which has been used as a Flink dynamic table to perform consequence processing after a lookup join
    • It is typically located in the middle layer of a data warehouse, between the raw data contained in the Operational Data Store (ODS) and the highly aggregated data found in the Data Warehouse (DW), or Data Mart (DM).
    • The primary purpose of the DWD layer is to support complex data analysis and reporting needs by providing a detailed and comprehensive data view.
    • Both the fact table and DWD table are hive tables on Hadoop
  3. Use a streaming cluster for the real-time processing. In this example, we use an Amazon EMR cluster to stream event ingestion and analyze it using Flink, using Flink Kafka connector and Hive connector to join the streaming event data and statics dimension data (fact table)

One of the key challenges encountered with this approach is related to the management of the lookup dimension table data. Initially, when the Flink application is started, this data is stored in the task manager’s state. However, during subsequent operations like continuous queries or window aggregations, the dimension table data isn’t automatically refreshed. This means that the operator must either restart the Flink application periodically or manually refresh the dimension table data in the temporary table. This step is crucial to ensure that the join operations and aggregations are always performed with the most current dimension data.

Another significant challenge with this approach is needing to pull the entire dimension table data and perform a cold start each time. This becomes particularly problematic when dealing with a large volume of dimension table data. For instance, when handling tables with tens of millions of registered users or tens of thousands of product SKU attributes, this process generates substantial input/output (IO) overhead. Consequently, it leads to performance bottlenecks, impacting the efficiency of the system.

Flink’s checkpointing mechanism processes the data and stores checkpoint snapshots of all the states during continuous queries or window aggregations, resulting in state snapshots data bloat.

Optimizing the solution

This post includes an optimized solution to address the aforementioned challenges, by automatically loading Hive dimension table data into the Alluxio UFS via the Alluxio cache layer. We join this data with Flink’s temporal joins to create a view on a changing table. This view reflects the content of a table at a specific point in time

Alluxio is a distributed cache engine for big data technology stacks. It provides a unified UFS that can connect to the underlying Amazon S3 and HDFS data. Alluxio UFS read and write operations warm up the distributed storage layers on S3 and HDFS and thus significantly increase throughput and reducing network overhead. Deeply integrated with upper level computing engines such as Hive, Spark, and Trino, Alluxio is an excellent cache accelerator for offline dimension data.

Additionally, we utilize Flink’s temporal table function to pass a time parameter. This function returns a view of the temporal table at the specified time. By doing so, when the main table of the real-time dynamic table is correlated with the temporal table, it can be associated with a specific historical version of the dimension data

Solution implementation details

For this post, we use “user behavior” log data in Kafka as real-time stream fact table data, and user information data on Hive as offline dimension table data. A demo with Alluxio + Flink temporal join is used to verify the Flink join optimized solution.

Real-time fact tables

For this demonstration, we utilize user behavior JSON data simulated by the open-source component json-data-generator. We write the data to Amazon Managed Kafka (Amazon MSK) in real-time. Using the Flink Kafka Connector, we convert this stream into a Flink stream table for continuous queries. This served as our fact table data for real-time joins.

A sample of the user behavior simulation data in JSON format is as follows:

[{          
	"timestamp": "nowTimestamp()",
	"system": "BADGE",
	"actor": "Agnew",
	"action": "EXIT",
	"objects": ["Building 1"],
	"location": "45.5,44.3",
	"message": "Exited Building 1"
}]

It includes user behavior information such as operation time, login system, user signature, behavioral activities, and service objects, locations, and related text fields. We create a fact table in Flink SQL with the main fields as follows:

CREATE TABLE logevent_source (`timestamp`  string, 
`system` string,
 actor STRING,
 action STRING
) WITH (
'connector' = 'kafka',
'topic' = 'logevent',
'properties.bootstrap.servers' = 'b-6.msk06.dr04w4.c3.kafka.ap-southeast-1.amazonaws.com:9092,b-5.msk06.dr04w4.c3.kafka.ap-southeast-1.amazonaws.com:9092,b-1.msk06.dr04w4.c3.kafka.ap-southeast-1.amazonaws.com:9092 (http://b-6.msk06.dr04w4.c3.kafka.ap-southeast-1.amazonaws.com:9092%2Cb-5.msk06.dr04w4.c3.kafka.ap-southeast-1.amazonaws.com:9092%2Cb-1.msk06.dr04w4.c3.kafka.ap-southeast-1.amazonaws.com:9092/)',
'properties.group.id' = 'testGroup6',
'scan.startup.mode'='latest-offset',
'format' = 'json'
);

Caching dimension tables with Alluxio

Amazon EMR provides solid integration with Alluxio. You can use the Amazon EMR bootstrap startup script to automatically deploy Alluxio components and start the Alluxio master and worker processes when an Amazon EMR cluster is created. For detailed installation and deployment steps, refer to the article Integrating Alluxio on Amazon EMR.

In an Amazon EMR cluster that integrates Alluxio, you may use Alluxio to create a cache table for the Hive offline dimension table as follows:

##Set up the client jar package in hive-env.sh:
$ export HIVE_AUX_JARS_PATH=/<PATH_TO_ALLUXIO>/client/alluxio-2.2.0-client.jar:${HIVE_AU

##Make sure the UFS is configured on the EMR cluster where Alluxio is installed and that the table/db path has been created:
alluxio fs mkdir alluxio://ip-xxx-xxx-xxx-xxx.ap-southeast-1.compute.internal:19998/s3/customer
alluxio fs chown hadoop:hadoop alluxio://ip-xxx-xxx-xxx-xxx.ap-southeast-1.compute.internal:19998/s3/customer

##On the AWS EMR cluster, create a Hive table path pointing to Alluxio namespace URI:
!connect jdbc:hive2://xxx.xxx.xxx.xxx:10000/default;
hive> CREATE TABLE customer(
    c_customer_sk             bigint,
    c_customer_id             string,
    c_current_cdemo_sk        bigint,
    c_current_hdemo_sk        bigint,
    c_current_addr_sk         bigint,
    c_first_shipto_date_sk    bigint,
    c_first_sales_date_sk     bigint,
    c_salutation              string,
    c_first_name              string,
    c_last_name               string,
    c_preferred_cust_flag     string,
    c_birth_day               int,
    c_birth_month             int,
    c_birth_year              int,
    c_birth_country           string,
    c_login                   string,
    c_email_address           string
)
    ROW FORMAT DELIMITED
    FIELDS TERMINATED BY '|'
    STORED AS TEXTFILE
    LOCATION 'alluxio://ip-xxx-xxx-xxx-xxx.ap-southeast-1.compute.internal:19998/s3/customer';
OK
Time taken: 3.485 seconds

As shown in the previous section, the Alluxio table location alluxio://ip-xxx-xx:19998/s3/customer points to the S3 path where the Hive dimension table is located; writing to the customer dimension table is automatically synchronized to the Alluxio cache.

After creating the Alluxio Hive offline dimension table, you can view the details of the Alluxio cache table by connecting to the Hive metadata through the Hive catalog in Flink SQL:

CREATE CATALOG hiveCatalog WITH (  'type' = 'hive',
    'default-database' = 'default',
    'hive-conf-dir' = '/etc/hive/conf/',
    'hive-version' = '3.1.2',
    'hadoop-conf-dir'='/etc/hadoop/conf/'
);
-- set the HiveCatalog as the current catalog of the session
USE CATALOG hiveCatalog;
show create table customer;
create external table customer(
    c_customer_sk             bigint,
    c_customer_id             string,
    c_current_cdemo_sk        bigint,
    c_current_hdemo_sk        bigint,
    c_current_addr_sk         bigint,
    c_first_shipto_date_sk    bigint,
    c_first_sales_date_sk     bigint,
    c_salutation              string,
    c_first_name              string,
    c_last_name               string,
    c_preferred_cust_flag     string,
    c_birth_day               int,
    c_birth_month             int,
    c_birth_year              int,
    c_birth_country           string,
    c_login                   string,
    c_email_address           string
) 
row format delimited fields terminated by '|'
location 'alluxio://ip-xxx-xxx-xxx-xxx.ap-southeast-1.compute.internal:19998/s3/30/customer' 
TBLPROPERTIES (
  'streaming-source.enable' = 'false',  
  'lookup.join.cache.ttl' = '12 h'
)

As shown in the preceding code, the location path of the dimension table is the UFS cache path Uniform Resource Identifier (URI). When the business program reads and writes the dimension table, Alluxio automatically updates the customer dimension table data in the cache and asynchronously writes it to the Alluxio backend storage path of the S3 table to achieve table data synchronization in the data lake.

Flink temporal table join

Flink temporal table is also a type of dynamic table. Each record in the temporal table is correlated with one or more time fields. When we join the fact table and the dimension table, we usually need to obtain real-time dimension table data for the lookup join. Thus, when creating or joining a table, we usually need to use the proctime() function to specify the time field of the fact table. When we join the tables, we use the syntax of FOR SYSTEM_TIME AS OF to specify the time version of the fact table that corresponds to the time of the lookup dimension table.

For this post, the customer information is a changing dimension table in the Hive offline table, whereas the customer behavior is the fact table in Kafka. We specified the time field with proctime() in the Flink Kafka source table. Then when joining the Flink Hive table, we used FOR SYSTEM_TIME AS OF to specify the time field of the lookup Kafka source table to allow us to realize the Flink temporal table join operation

As shown in the following code, a fact table of user behavior is created through the Kafka Connector in Flink SQL. The ts field refers to the timestamp when the temporal table is joined：

CREATE TABLE logevent_source (`timestamp`  string, 
`system` string,
 actor STRING,
 action STRING,
 ts as PROCTIME()
) WITH (
'connector' = 'kafka',
'topic' = 'logevent',
'properties.bootstrap.servers' = 'b-6.msk06.dr04w4.c3.kafka.ap-southeast-1.amazonaws.com:9092,b-5.msk06.dr04w4.c3.kafka.ap-southeast-1.amazonaws.com:9092,b-1.msk06.dr04w4.c3.kafka.ap-southeast-1.amazonaws.com:9092 (http://b-6.msk06.dr04w4.c3.kafka.ap-southeast-1.amazonaws.com:9092%2Cb-5.msk06.dr04w4.c3.kafka.ap-southeast-1.amazonaws.com:9092%2Cb-1.msk06.dr04w4.c3.kafka.ap-southeast-1.amazonaws.com:9092/)',
'properties.group.id' = 'testGroup-01',
'scan.startup.mode'='latest-offset',
'format' = 'json'
);

The Flink offline dimension table and the streaming real-time table are joined as follows:

select a.`timestamp`,a.`system`,a.actor,a.action,b.c_login from 
       (select *, proctime() as proctime from user_logevent_source) as a 
 left join customer  FOR SYSTEM_TIME AS OF a.proctime as b on a.actor=b.c_last_name;

When the fact table logevent_source joins the lookup dimension table, the proctime function ensures real-time joins by obtaining the latest dimension table version. This dimension data, cached in Alluxio, delivers significantly better read performance than direct S3 access.

At the same time, the dimension table data is already cached in Alluxio; the read performance is much better than offline data read on S3.

The comparison test shows that Alluxio cache brings a clear performance advantage by switching the S3 and Alluxio paths of the customer dimension table through Hive

You can easily switch the local and cache location paths with alter table in hive cli:

alter table customer set location "s3://xxxxxx/data/s3/30/customer";
alter table customer  set location "alluxio://ip-xxx-xxx-xxx-xxx.ap-southeast-1.compute.internal:19998/s3/30/customer";

You can also select the Task Manager log from the Flink dashboard for a split test.

The performance of the fact table load was doubled through the implementation of optimized data processing techniques.

  1. Before caching (S3 path read): 5s load time 
    2022-06-29 02:54:34,791 INFO  com.amazon.ws.emr.hadoop.fs.s3n.S3NativeFileSystem           [] - Opening 's3://salunchbucket/data/s3/30/customer/data-m-00029' for reading
2022-06-29 02:54:39,971 INFO  org.apache.flink.table.filesystem.FileSystemLookupFunction   [] - Loaded 433000 row(s) into lookup join cache

  2. After caching (Alluxio read): 2s load time 
    2022-06-29 03:25:14,476 INFO  com.amazon.ws.emr.hadoop.fs.s3n.S3NativeFileSystem           [] - Opening 's3://salunchbucket/data/s3/30/customer/data-m-00029' for reading
2022-06-29 03:25:16,397 INFO  org.apache.flink.table.filesystem.FileSystemLookupFunction   [] - Loaded 433000 row(s) into lookup join cache

The timeline on JobManager clearly shows the difference in execution duration under Alluxio and S3 paths.

For single task query ,we accelerate by more than 1 times using this solution. The overall job performance improvement is even more visible.

Other optimalizations to consider

Implementing a continuous join requires pulling dimension data every time. Does it lead to Flink’s checkpoint state bloat that can cause Flink TaskManager RocksDB to explode or memory overflow.
In Flink, the state comes with a TTL mechanism. You can set a TTL expiration policy to trigger Flink to clean up expired state data. Flink SQL can be set using the hint method.

insert into logevent_sink
select a.`timestamp`,a.`system`,a.actor,a.action,b.c_login from 
(select *, proctime() as proctime from logevent_source) as a 
  left join 
customer/*+ OPTIONS('lookup.join.cache.ttl' = '5 min')*/  FOR SYSTEM_TIME AS OF a.proctime as b 
on a.actor=b.c_last_name;

Flink Table/Streaming API is similar:

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.days(7))
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .cleanupInRocksdbCompactFilter() 
    .build();
ValueStateDescriptor<Long> lastUserLogin = 
    new ValueStateDescriptor<>("lastUserLogin", Long.class);
lastUserLogin.enableTimeToLive(ttlConfig);
StreamTableEnvironment.getConfig().setIdleStateRetentionTime(min, max);

Restart the lookup join after the configuration. As you can see from the Flink TM log, after TTL expires, it triggers clean-up and re-pull the Hive dimension table data：

2022-06-29 04:17:09,161 INFO  org.apache.flink.table.filesystem.FileSystemLookupFunction   
[] - Lookup join cache has expired after 5 minute(s), reloading

In addition, you can reduce the number of checkpoint snapshots by configuring Flink state retention and thereby reduce the amount of space taken up by state at the time of snapshot.

Flink job configuration as follow：
-D state.checkpoints.num-retained=5 

After the configuration, you can see that in the S3 checkpoint path, the Flink job automatically cleans up historical snapshots and keeps the most recent 5 snapshots, thus ensuring that checkpoint snapshots do not accumulate.

[hadoop@ip-172-31-41-131 ~]$ aws s3 ls s3://salunchbucket/data/checkpoints/7b9f2f9becbf3c879cd1e5f38c6239f8/
                           PRE chk-3/
                           PRE chk-4/
                           PRE chk-5/
                           PRE chk-6/
                           PRE chk-7/

Summary

Customers implementing Flink streaming framework to join dimension and real-time fact tables frequently encounter performance challenges. In this post, we presented an optimized solution that uses Alluxio’s caching capabilities to automatically load Hive dimension table data into the UFS cache. By integrating with Flink temporal table joins, dimension tables are transformed into time-versioned views, effectively addressing performance bottlenecks in traditional implementations.

About the author

Jeff Tang

Jeff Tang

Jeff is a Data Analytics Solutions Architect at AWS. He’s responsible for designing and optimizing Amazon Data Analytic services, with over 10 years of experience in data architecture and development. Former roles include Senior Consulting Advisor at Oracle, Senior Architect at Migu Culture Data Market, and Data Analytics Architect at ANZ Bank. Extensive experience in big data, data lakes, intelligent lakehouses, and MLOps platforms

The collective thoughts of the interwebz