Coding

Columnar storage was introduced in SQL Server 2016 as part of the SQL Server 2016 In-Memory OLTP feature. It is specifically designed for data warehousing and analytical workloads, where large amounts of data need to be scanned, aggregated, or analyzed efficiently. Columnar storage stores data in a column-wise format rather than the traditional row-wise storage, offering significant performance benefits for read-heavy operations such as reporting and analytics. Key Benefits of Columnar Storage Faster read performance: Optimized for analytics where only a few columns are needed in a query. Compression: Since column data is homogeneous, it achieves high compression rates, saving storage space. Improved query performance: Aggregating or scanning specific columns is much faster in a columnar format, especially with large datasets. Setting Up Columnar Tables in SQL Server SQL Server implements columnar storage through the Columnstore Index. The Columnstore Index is a special kind of index used in large data tables where the data is stored in columns rather than rows. The clustered columnstore index (CCI) is the preferred method when creating columnar tables. Step 1: Create a Sample Table Let's start by creating a table with a large number of rows, which we will populate with random data to demonstrate the difference between row-store and column-store formats. SQL -- Creating a traditional Rowstore Table CREATE TABLE SalesData_RowStore ( SalesOrderID INT, ProductID INT, Quantity INT, SalesAmount DECIMAL(18, 2), OrderDate DATE ); Step 2: Insert Data Into Rowstore Table For the sake of performance demonstration, we will generate a large set of random data. MS SQL -- Generate a large set of random data for Rowstore Table DECLARE @Counter INT = 0; WHILE @Counter < 1000000 BEGIN INSERT INTO SalesData_RowStore (SalesOrderID, ProductID, Quantity, SalesAmount, OrderDate) VALUES (FLOOR(RAND() * 1000) + 1, FLOOR(RAND() * 100) + 1, FLOOR(RAND() * 100) + 1, FLOOR(RAND() * 500) + 1, DATEADD(DAY, FLOOR(RAND() * 365) + 1, GETDATE())); SET @Counter = @Counter + 1; END Implementing Columnstore Index (Columnar Table) Step 1: Create a Columnstore Table Now, let's create a table with a clustered columnstore index (CCI). This index allows SQL Server to store the data in a columnar format. MS SQL -- Creating a Columnstore Table with Clustered Columnstore Index CREATE TABLE SalesData_ColumnStore ( SalesOrderID INT, ProductID INT, Quantity INT, SalesAmount DECIMAL(18, 2), OrderDate DATE ); MS SQL CREATE CLUSTERED COLUMNSTORE INDEX CCI_SalesData ON SalesData_ColumnStore; Step 2: Insert the Same Data Into the Columnstore Table You can insert the same large dataset into the columnar table in the same way. SQL -- Insert data into Columnstore Table DECLARE @Counter INT = 0; WHILE @Counter < 1000000 BEGIN INSERT INTO SalesData_ColumnStore (SalesOrderID, ProductID, Quantity, SalesAmount, OrderDate) VALUES (FLOOR(RAND() * 1000) + 1, FLOOR(RAND() * 100) + 1, FLOOR(RAND() * 100) + 1, FLOOR(RAND() * 500) + 1, DATEADD(DAY, FLOOR(RAND() * 365) + 1, GETDATE())); SET @Counter = @Counter + 1; END Query Performance Without Columnar Index Let's execute a typical query that aggregates data by ProductID and OrderDate. This will involve scanning through a large amount of data in the rowstore table. MS SQL -- Query on Rowstore Table SELECT ProductID, SUM(SalesAmount) AS TotalSales FROM SalesData_RowStore WHERE OrderDate > '2023-01-01' GROUP BY ProductID; Expected Outcome The query will scan all the rows in the table. Rowstore tables are not optimized for this type of query, and the performance might degrade with large datasets due to the need to read each row. Query Performance With Columnar Index Let's run the same query on the columnar table using a Clustered Columnstore Index. MS SQL -- Query on Columnstore Table SELECT ProductID, SUM(SalesAmount) AS TotalSales FROM SalesData_ColumnStore WHERE OrderDate > '2023-01-01' GROUP BY ProductID; Expected Outcome The columnar index stores the data by columns, and SQL Server can read only the relevant columns for the query (i.e., ProductID and SalesAmount). Columnstore indexes are highly optimized for these types of queries, resulting in much faster query execution time. Comparing the Performance of Both Scenarios To compare the performance of the two scenarios, we will execute both queries and check the execution plan and query duration. Step 1: Query Execution Plan Without Columnstore You can use the following query to view the execution plan for the rowstore table. MS SQL -- Displaying Execution Plan for Rowstore Table SET STATISTICS IO ON; SET STATISTICS TIME ON; SELECT ProductID, SUM(SalesAmount) AS TotalSales FROM SalesData_RowStore WHERE OrderDate > '2023-01-01' GROUP BY ProductID; SET STATISTICS IO OFF; SET STATISTICS TIME OFF; This will provide information on: Logical reads: The number of data pages read from diskCPU time: How much CPU time was consumedElapsed time: The total time taken to execute the query Step 2: Query Execution Plan With Columnstore Now, execute the same for the columnstore table. MS SQL -- Displaying Execution Plan for Columnstore Table SET STATISTICS IO ON; SET STATISTICS TIME ON; SELECT ProductID, SUM(SalesAmount) AS TotalSales FROM SalesData_ColumnStore WHERE OrderDate > '2023-01-01' GROUP BY ProductID; SET STATISTICS IO OFF; SET STATISTICS TIME OFF; In the execution plan for the columnstore table, SQL Server will typically show fewer logical reads and significantly lower CPU time, as it only scans the necessary columns. Performance Improvements in Columnar Tables Scenario 1: Data Compression Columnar storage achieves higher compression rates because data is stored in homogeneous chunks, which makes it more efficient in terms of storage. Compression reduces disk I/O during query execution. Scenario 2: Selective Column Scanning When querying only a few columns, columnar storage avoids scanning the entire row. In contrast, rowstore requires scanning all columns in every row, even if only a subset is required for the query. Conclusion In this example, we demonstrated how implementing columnstore indexes in SQL Server can significantly improve query performance, especially for analytics and aggregation queries on large datasets. The comparison showed that columnar storage excels in reducing query times by optimizing disk I/O, leveraging data compression, and selectively reading only the necessary columns. As a result, columnstore indexing is a great choice for data warehousing or any scenario where read performance for large datasets is critical.

By arvind toorpu CORE

Most agent demos work beautifully on stage and fall apart the first week in production. The reason is almost always the same: the demo treats tool-calling as a happy path, and production is nothing but edge cases. A tool times out. A model hallucinates an argument. The agent loops on itself and burns through your token budget. After shipping a few of these systems, I have learned that the durable design question is not "can the agent call a tool" but "what happens when the tool call goes wrong." This tutorial walks through a tool-calling agent in LangGraph built the way I would build it for production, with the safeguards baked in from the first commit rather than bolted on after the first incident. What We Are Building To keep the focus on production patterns rather than business logic, we will build a small but realistic agent: a currency assistant. A user asks a plain-language question like "What is the USD to INR rate?" and the agent answers using live foreign-exchange data rather than guessing. The model itself has no idea what today's rate is, so it must recognize that it needs data, call a get_exchange_rate tool to fetch it, and then return the actual result. That is the entire reason tool-calling exists: it turns a model that can only talk into an agent that can act and ground its answers in real data. FX rates are a good teaching example because the failure modes are obvious and unforgiving. A wrong currency code, an unsupported pair, or a flaky data source are exactly the kinds of things that must not crash a production agent. The Mental Model A tool-calling agent is a loop. The model looks at the conversation and decides whether to answer directly or to call a tool. If it calls a tool, your code runs that tool, feeds the result back, and the model decides again. That loop is exactly what LangGraph is good at expressing: nodes do work, edges decide where control flows next. Step 1: A Tool That Cannot Crash Your Agent Our agent's one capability is looking up an exchange rate, so the tool that does it has to be bulletproof. The single most important production habit is that a tool never raises into the agent loop. It validates its input and returns a readable error string that the model can reason about. A raised exception kills the run; a returned error lets the agent recover. Here, the tool checks that both currency codes are valid and that the requested pair exists, returning a clear message when either check fails. Notice the failure modes are first-class outputs, not afterthoughts. Bad arguments and missing data both produce a controlled message. Step 2: The Nodes The agent node asks the model what to do. The tool node executes any requested tools, and critically, it wraps every call so a failure becomes a message instead of a stack trace. It also rejects calls to tools that do not exist, which is how you contain a hallucinated tool name. Step 3: Wiring the Loop, With a Brake The conditional edge sends control to the tools node only when the model actually requests a tool; otherwise, the run ends. This is the whole agent loop in four lines. The brake that separates a demo from a production system is one line at invocation time: The recursion_limit bounds how many times the loop can cycle. Without it, a confused model can call tools indefinitely. With it, a runaway agent fails fast and loudly instead of quietly draining your budget. Treat it as a required parameter, not an optional one. Step 4: Observability, You Will Thank Yourself For When an agent misbehaves in production, you need to see its decisions, not guess at them. A few log lines at each node turn an opaque black box into a traceable sequence. Running the agent against "What is the USD to INR rate?" produces this: Every step is visible: the agent decides to call a tool, the tool returns a validated result, control flows back, and the model produces its final answer. When something breaks at 2 a.m., this trace is the difference between a five-minute fix and a five-hour investigation. What Makes It Survive The example is small, but the principles scale. Tools return errors instead of raising. Unknown tool names are rejected rather than executed. The loop is bounded, so it cannot run away. Every decision is logged, so failures are traceable instead of mysterious when you are debugging at scale. Swap the prototype model for ChatAnthropic or ChatOpenAI, add your real tools, and the same skeleton carries you from prototype to production without rewriting the core. The hard part of agent engineering was never getting the model to call a tool. It is designed for the moment the call goes wrong, and LangGraph gives you exactly the right place to put each safeguard.

HTTP methods in REST API design are more than technical details; they communicate intent between clients and servers. A GET request instructs the server to retrieve a resource. A POST request typically indicates that data should be processed, often creating a new resource. PUT indicates replacement or update, while DELETE signals removal. These methods are well-established and fundamental to the Web. Despite this, API design has long faced a notable gap. Challenges arise when a client needs to retrieve data using queries too complex for a URL. Filters such as destination, price range, availability, category, user preferences, pagination, sorting, and business rules can be added as query parameters, but this often results in lengthy, hard-to-read URLs that are difficult to maintain and may not be suitable for sensitive or structured data. For years, the common workaround was to use POST for search operations: HTTP POST /trips/search Content-Type: application/json However, this approach does not align with HTTP semantics. Searching is typically a safe operation that does not alter server state and is often idempotent, producing the same result if the underlying data remains unchanged. POST does not clearly convey this intent, and can complicate caching and retries, and the API documentation is less precise. The new HTTP QUERY method addresses this specific need. The QUERY method provides a dedicated way to send structured request content while indicating that the operation is safe and idempotent. It functions similarly to GET but allows the client to include a request body, as with POST. According to RFC 10008, a QUERY request asks the target resource to process the enclosed content safely and idempotently, then return the result. This matters because modern APIs are no longer limited to. This is important because modern APIs often require more than simple resource retrieval. Use cases such as search screens, dashboards, reporting APIs, recommendation engines, GraphQL-like endpoints, analytics filters, and domain-specific query languages demand more expressive input than URLs can provide, yet do not represent state-changing operations. cation protocol, not merely as a transport tunnel. The word “method” in HTTP is important: it defines the request's semantic purpose. QUERY continues that tradition by giving read-oriented complex operations their own explicit place in the protocol. This article will examine the purpose of QUERY, its differences from GET and POST, appropriate use cases, and its potential to enhance modern Java REST API design. Building the Sample Project With the importance of the HTTP QUERY method established, let’s transition from concept to implementation. This sample uses a travel agency domain. The goal is to expose a search operation that allows clients to filter travel offers by city, travel type, and price range. In such cases, a traditional GET URL can become cumbersome, while using POST is not semantically appropriate. The project uses Helidon 4.5.0, generated from the official Helidon Starter. Helidon MP provides a MicroProfile/Jakarta-oriented programming model suitable for this example. After generating the project, configure the dependencies. This sample uses Eclipse JNoSQL 1.2.0-M1 to demonstrate Jakarta NoSQL and Jakarta Data integration. XML <properties> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <project.report.sourceEncoding>UTF-8</project.report.sourceEncoding> <maven.compiler.release>21</maven.compiler.release> <jnosql.version>1.2.0-M1</jnosql.version> </properties> <dependencies> <dependency> <groupId>org.eclipse.jnosql.databases</groupId> <artifactId>jnosql-oracle-nosql</artifactId> <version>${jnosql.version}</version> </dependency> <dependency> <groupId>org.eclipse.jnosql.metamodel</groupId> <artifactId>mapping-metamodel-processor</artifactId> <version>${jnosql.version}</version> <scope>provided</scope> </dependency> </dependencies> Two key dependencies are included. The first dependency enables Eclipse JNoSQL integration with Oracle NoSQL, which supports key-value and document-oriented models. This example uses the document model, as the travel entity maps naturally to a document structure. The second dependency enables the metamodel processor, which generates the _Travel class. This allows us to use type-safe Jakarta Data restrictions, such as _Travel.city.equalTo(city), for fluent and dynamic queries. Creating the Domain Entity With dependencies configured, we can define the domain model. The Travel entity represents each travel offer, including an identifier, destination city, travel type, and price. Jakarta NoSQL supports entities as regular classes or Java records. For this small, immutable sample, a Java record is appropriate. Java import jakarta.nosql.Column; import jakarta.nosql.Entity; import jakarta.nosql.Id; import java.math.BigDecimal; import java.util.UUID; @Entity public record Travel( @Id UUID id, @Column String city, @Column TravelType type, @Column BigDecimal price) { } These annotations are similar to those in Jakarta Persistence. While the vocabulary differs, the intent remains familiar. Java import jakarta.nosql.Column; import jakarta.nosql.Entity; import jakarta.nosql.Id; @Entity marks the record as persistent. @Id specifies the primary identifier. @Column maps fields to the NoSQL database. Now we define the travel category: Java public enum TravelType { BUSINESS, LEISURE } This approach provides a compact and expressive domain model for the remainder of the article. Creating the Repository With Jakarta Data The next step is to create the bridge between Java and the database. We use Jakarta Data, which offers a repository-oriented programming model compatible with various persistence technologies. In this sample, the repository uses Eclipse JNoSQL and Oracle NoSQL, while the programming model remains domain-focused. Java import jakarta.data.repository.BasicRepository; import jakarta.data.repository.Find; import jakarta.data.repository.Repository; import jakarta.data.restrict.Restriction; import java.util.List; import java.util.UUID; @Repository public interface TravelRepository extends BasicRepository<Travel, UUID> { @Find List<Travel> query(Restriction<Travel> restriction); default boolean isEmpty() { return countBy() == 0; } long countBy(); } The repository extends BasicRepository<Travel, UUID>, which gives us common persistence operations such as saving entities and retrieving data. The key method for this article is: Java @Find List<Travel> query(Restriction<Travel> restriction); A Restriction<Travel> represents a dynamic query condition, which aligns well with the HTTP QUERY scenario. Clients can send various combinations of filters, such as city, price, travel type, or all of them. Instead of creating separate repository methods for each combination, we build queries dynamically. The isEmpty() method is a convenience for loading initial data at application startup: Java default boolean isEmpty() { return countBy() == 0; } long countBy(); Creating the Filter Request DTO Next, create the object that represents the request. Here, the HTTP QUERY method is useful. Instead of encoding each filter in the URL, we send a structured request body. The Java representation is as follows: Java import expert.os.demos.travel.infrastructure.FieldVisibilityStrategy; import jakarta.json.bind.annotation.JsonbVisibility; import java.math.BigDecimal; import java.util.Optional; @JsonbVisibility(value = FieldVisibilityStrategy.class) public class TravelFilterRequest { private String city; private TravelType type; private BigDecimal minPrice; private BigDecimal maxPrice; public Optional<String> city() { return Optional.ofNullable(city); } public Optional<TravelType> type() { return Optional.ofNullable(type); } public Optional<BigDecimal> minPrice() { return Optional.ofNullable(minPrice); } public Optional<BigDecimal> maxPrice() { return Optional.ofNullable(maxPrice); } @Override public String toString() { return "TravelRequest{" + "city='" + city + '\'' + ", type=" + type + ", minPrice=" + minPrice + ", maxPrice=" + maxPrice + '}'; } } A key design choice is that each accessor returns an Optional. This approach makes the filtering logic explicit. Fields may or may not be present in the request, so the service applies each restriction only when a value exists, avoiding null checks. The @JsonbVisibility annotation customizes how JSON-B accesses fields. Here, the DTO keeps fields private and exposes domain-friendly accessor methods such as city(), type(), minPrice(), and maxPrice(). Creating the Travel Service Now, we can create the service layer. In this sample, the service has two responsibilities: First, it loads initial travel data if the repository is empty. Second, it converts incoming filter requests into Jakarta Data restrictions. Java import jakarta.annotation.PostConstruct; import jakarta.data.restrict.Restrict; import jakarta.data.restrict.Restriction; import jakarta.enterprise.context.ApplicationScoped; import jakarta.inject.Inject; import org.eclipse.jnosql.mapping.Database; import org.eclipse.jnosql.mapping.DatabaseType; import java.util.ArrayList; import java.util.List; import java.util.UUID; import java.util.logging.Logger; @ApplicationScoped public class TravelService { private static final Logger LOGGER = Logger.getLogger(TravelService.class.getName()); private final TravelRepository travelRepository; @Inject public TravelService(@Database(DatabaseType.DOCUMENT) TravelRepository travelRepository) { this.travelRepository = travelRepository; } TravelService() { this.travelRepository = null; } @PostConstruct void load() { if (travelRepository.isEmpty()) { LOGGER.info("[TRAVEL SERVICE] Loading initial travel data..."); travelRepository.save(new Travel( UUID.randomUUID(), "New York", TravelType.BUSINESS, new java.math.BigDecimal("1500.00"))); travelRepository.save(new Travel( UUID.randomUUID(), "Paris", TravelType.LEISURE, new java.math.BigDecimal("2000.00"))); travelRepository.save(new Travel( UUID.randomUUID(), "Tokyo", TravelType.BUSINESS, new java.math.BigDecimal("3000.00"))); travelRepository.save(new Travel( UUID.randomUUID(), "Sydney", TravelType.LEISURE, new java.math.BigDecimal("1800.00"))); travelRepository.save(new Travel( UUID.randomUUID(), "Rome", TravelType.LEISURE, new java.math.BigDecimal("2500.00"))); } else { LOGGER.info("[TRAVEL SERVICE] Travel data already loaded."); } } public List<Travel> search(TravelFilterRequest filter) { LOGGER.info("[TRAVEL SERVICE] Searching for travels with filter: " + filter); if (filter == null) { return travelRepository.findAll().toList(); } List<Restriction<Travel>> restrictions = new ArrayList<>(); filter.city() .ifPresent(city -> restrictions.add(_Travel.city.equalTo(city))); filter.type() .ifPresent(type -> restrictions.add(_Travel.type.equalTo(type))); filter.minPrice() .ifPresent(minPrice -> restrictions.add(_Travel.price.greaterThanEqual(minPrice))); filter.maxPrice() .ifPresent(maxPrice -> restrictions.add(_Travel.price.lessThanEqual(maxPrice))); return travelRepository.query( Restrict.all(restrictions.toArray(new Restriction[0]))); } } Enabling the HTTP QUERY Method With JAX-RS The final step is to enable the HTTP QUERY method in the Java API. JAX-RS simplifies this process. The specification allows custom HTTP method annotations using @HttpMethod. Since QUERY is now an HTTP method, we can create a concise annotation and apply it directly in the resource class. Java import jakarta.ws.rs.HttpMethod; import java.lang.annotation.ElementType; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; @Target({ElementType.METHOD}) @Retention(RetentionPolicy.RUNTIME) @HttpMethod("QUERY") public @interface QUERY { } This annotation functions similarly to the built-in JAX-RS annotations such as @GET, @POST, @PUT, and @DELETE. The key line is: Java @HttpMethod("QUERY") This informs JAX-RS that methods annotated with @QUERY handle incoming HTTP requests using the QUERY method. At this stage, the API expresses the operation more clearly. POST is no longer used as a workaround for search. This endpoint now receives a query payload and returns a result without altering server state. Exposing the Travel Resource We can now expose the resource. Java package expert.os.demos.travel; import expert.os.demos.travel.infrastructure.QUERY; import jakarta.enterprise.context.ApplicationScoped; import jakarta.inject.Inject; import jakarta.ws.rs.Consumes; import jakarta.ws.rs.Path; import jakarta.ws.rs.Produces; import jakarta.ws.rs.core.MediaType; import java.util.List; import java.util.logging.Logger; @ApplicationScoped @Consumes(MediaType.APPLICATION_JSON) @Produces(MediaType.APPLICATION_JSON) @Path("/travels") public class TravelResource { private static final Logger LOGGER = Logger.getLogger(TravelResource.class.getName()); @Inject private TravelService travelService; @QUERY public List<Travel> search(TravelFilterRequest filter) { LOGGER.info("Searching travels with filter: " + filter); return travelService.search(filter); } } The resource is intentionally small. Its job is not to understand database details or build dynamic queries. It only receives the HTTP request, delegates the filter to the service, and returns the result. Java @QUERY public List<Travel> search(TravelFilterRequest filter) { LOGGER.info("Searching travels with filter: " + filter); return travelService.search(filter); } This method is the article's focal point. Executing the Request With the service running, test the endpoint using a client that supports custom HTTP methods. For example, in Postman: HTTP QUERY http://localhost:8081/travels Content-Type: application/json { "type": "BUSINESS", "maxPrice": 2800 } Alternatively, in Postman-style command form: Shell postman request QUERY 'http://localhost:8081/travels' \ --header 'Content-Type: application/json' \ --body '{"type": "BUSINESS", "maxPrice": 2800}' Given the initial data loaded by the service, this request retrieves business trips with a price of 2800 or less. The matching result should include New York: JSON [ { "id": "generated-uuid", "city": "New York", "type": "BUSINESS", "price": 1500.00 } ] Tokyo is also a business trip, but its price is 3000.00, so it does not match the maxPrice filter. This example demonstrates the value of QUERY: the client can send a structured request body, the server preserves read-oriented semantics, and the backend maps the request directly into a dynamic Jakarta Data query.

By Otavio Santana CORE

In modern application development, feature flags are the guardrails that keep experiments controlled and rollbacks safe when conditions shift. If feature flags act as the guardrails, observability provides the visibility: the headlights (traces), mirrors (logs), and dashboard instruments (metrics) that reveal what’s happening in the environment and how well a feature is performing. Together, feature flags and observability unlock powerful insights by correlating code changes with real-time system behavior. This combination reduces time-to-diagnosis and builds greater confidence when rolling out new features. In this post, we’ll walk through just how to add observability to a React Native application using LaunchDarkly’s observability SDK. To demonstrate the process, we’ll build on the PlusOne app, a simple counter app that includes increment (+1), reset, and error-triggering buttons. This lightweight demo provides a clean foundation to showcase how logs, traces, and errors can seamlessly flow into LaunchDarkly for monitoring and debugging. Prerequisites LaunchDarkly account. Sign up for a free one here.Visual Studio or another code editor of choice. All code from this tutorial can be found on GitHub. Setting Up Your Environment Before running a React Native app, make sure your development environment is set up correctly. You can find the full setup instructions for both Android and iOS here. In this tutorial, we'll be running iOS, but keep in mind Expo Orbit, the platform we'll be using to run our iOS simulator, requires both Xcode and Android Studio to be installed. After going through the instructions, you should have the following installed: Node JS (preferably via nvm)Watchman for file monitoringJDK via zulu package managerAndroid Studio. Don’t forget to set your Android_Home environment variablesXcode for the iOS simulatorCocoapods for iOS dependency managementExpo Orbit for running Expo apps on Android or iOS If you're using Android, don't forget to add your environment variables to bash or zsh profile. JavaScript export ANDROID_HOME=$HOME/Library/Android/sdk export PATH=$PATH:$ANDROID_HOME/emulator export PATH=$PATH:$ANDROID_HOME/platform-tools Starting Up the PlusOne App To get started, let’s clone the repo for the PlusOne app and run npm install to ensure the proper dependencies are present in our node_modules file. Clone the repo. JavaScript git clone https://github.com/arober39/PlusOne Install dependencies using npm. JavaScript cd PlusOne npm install We’ll also need to run both the prebuild command to generate the iOS file and the expo run command to run the iOS simulator. Prebuild for iOS. JavaScript npx expo prebuild Run expo app. JavaScript npm expo run:ios Now we can view the iOS app in the iPhone simulator using npm. JavaScript # iOS npm run ios # Android npm run android The app should look something like this: Feel free to interact with the app to ensure all is working as expected. As you can see in the code, we have three buttons: one that adds one to the displayed number, one to bring the count back to zero, and an intentional Error button to test error monitoring within the LaunchDarkly UI. JavaScript // app/index.tsx import { useState } from "react"; import { StyleSheet, Text, TouchableOpacity, View } from "react-native"; export default function Index() { const [count, setCount] = useState(0); const handleReset = () => setCount(0); const handleIncrement = () => setCount((prev) => prev + 1); const triggerRecordedError = () => { try { throw new Error("Simulated controlled error from Plus One app") } catch (e) { alert("You intentionally threw an error") } }; return ( <View style={styles.container}> <View style={styles.header}> <Text style={styles.headerText}>Plus One</Text> </View> <View style={styles.counterWrapper}> <Text style={styles.counterText}>{count}</Text> </View> <View style={styles.actionsRow}> <ButtonBox label="Reset" onPress={handleReset} /> <ButtonBox label="+1" onPress={handleIncrement} /> <ButtonBox label="Error" onPress={triggerRecordedError} /> </View> </View> ); } type ButtonBoxProps = { label: string; onPress: () => void; }; function ButtonBox({ label, onPress }: ButtonBoxProps) { return ( <TouchableOpacity onPress={onPress} style={styles.button} activeOpacity={0.8}> <Text style={styles.buttonText}>{label}</Text> </TouchableOpacity> ); } /* The rest of the application code */ Now that we have verified a working app, we can add observability support by downloading the observability React Native SDK. Install LaunchDarkly SDK dependencies. JavaScript npm install @launchdarkly/react-native-client-sdk npm install @launchdarkly/observability-react-native Next, you’ll need to initialize the React Native LD client in the app/_layout file. Replace the in the layout file by pasting the following code. JavaScript // app/_layout.tsx import { Observability } from '@launchdarkly/observability-react-native'; import { AutoEnvAttributes, LDOptions, LDProvider, ReactNativeLDClient } from '@launchdarkly/react-native-client-sdk'; import { Stack } from 'expo-router'; import { useEffect, useState } from 'react'; const options: LDOptions = { applicationInfo: { id: 'Plus-One', name: 'Sample Application', version: '1.0.0', versionName: 'v1', }, debug: true, plugins: [ new Observability({ serviceName: 'my-react-native-app', serviceVersion: '1.0.0', }) ], }; const userContext = { kind: 'user', key: 'test-hello' }; export default function RootLayout() { const [client, setClient] = useState<ReactNativeLDClient | null>(null); useEffect(() => { // Initialize client const featureClient = new ReactNativeLDClient( 'mob-abc123', AutoEnvAttributes.Enabled, options, ); featureClient.identify(userContext).catch((e: any) => console.log(e)); setClient(featureClient); // Cleanup function that runs when component unmounts return () => { featureClient.close(); }; }, []); if (!client) { return null; } return ( <LDProvider client={client}> <Stack /> </LDProvider> ); } First, we’re importing the Observability SDK as well as a few LD libraries to add options and attributes to the LD client. Initialized the SDK and plugin options.Defined the user context.Lastly, you initialized the client. Now that you have defined your LD React Native client, you can implement different observability methods within your application logic. We can do this by importing the LDObserve library in the app/_layout.tsx file. JavaScript import { LDObserve } from '@launchdarkly/observability-react-native'; Then, add the recordError() method within the triggerRecordedError function inside the app/_layout.tsx file. This will allow for error messages to be sent back to the LD UI. JavaScript const triggerRecordedError = () => { try { throw new Error("Simulated controlled error from Plus One app") } catch (e) { LDObserve.recordError(e as Error, {feature: "test-button"}) alert("You intentionally threw an error") } }; Before being able to receive data in the LD UI, you’ll need to add your mobile key to the React Native LD client, which can be found by logging in to the LD UI. Once logged in, tap the settings button at the bottom left. Navigate to the Projects page and click Create to create a new project. Define the new Project and click Create Project. Then, define the environment where you would like your data to be sent. Now, grab the mobile key by pressing the three dots for the environment and selecting the mobile key, which will copy the key to your keyboard. Then, add it to the app/_layout file. JavaScript const featureClient = new ReactNativeLDClient( ‘mob-abc123’, AutoEnvAttributes.Enabled, options, ); Finally, you can generate data by interacting with your app in the iOS app simulator. Feel free to restart the app to ensure data is displaying in real time. JavaScript npm expo run:ios Once you navigate back to the LD UI, you should be able to see the logs, traces, and errors under the Monitor section. Logs Traces Errors Conclusion In just a few minutes, we’ve taken the PlusOne React Native app from a simple counter to a fully observable application connected to LaunchDarkly. By setting up the SDK, initializing observability plugins, and recording errors, we now have a live feedback loop where application behavior is visible in the LaunchDarkly UI. This makes it far easier to diagnose issues, validate feature flag rollouts, and ensure smooth user experiences. Next Steps Looking ahead, there are many ways to expand on what we’ve built by including features like recording custom metrics and session replay, which provide even deeper insights into app behavior. By integrating observability at the foundation of your React Native projects, you equip your team with the clarity needed to debug faster, ship features more confidently, and deliver reliable experiences to your users. You can also read this article to learn more about observability and guarded releases.

Raw data doesn't win model competitions. Features do. And when your raw data is tens of billions of rows sitting across multiple sources, you can't afford to run pandas in a notebook and call it a day. In this tutorial, I'll walk through building a production-grade feature engineering pipeline on Azure Databricks using: Apache Spark for distributed transformation at scaleDelta Lake for reliable, versioned feature storage with ACID guaranteesMLflow for tracking feature pipeline runs, parameters, and the models trained on top of them The use case is a customer churn prediction system, but the patterns apply to any ML feature pipeline. Architecture Overview The pipeline follows the Medallion Architecture — a layered approach where data gets progressively cleaner and more feature-ready as it moves from Bronze to Silver to Gold. MLflow sits across all three layers, tracking every run. Pipeline Flow Layer Breakdown LayerDelta TableWhat happens hereTypical latencyBronzechurn.bronze.eventsRaw ingest, no transforms, append onlyMinutesSilverchurn.silver.customersDeduplication, null handling, schema enforcementMinutesGoldchurn.gold.featuresAggregations, window functions, encodingMinutes to hoursMLflow RunN/ATraining, metric logging, artifact storageHoursRegistryN/AVersioned model store, stage promotionOn demand Step 1 — Bronze Layer: Raw Ingest The Bronze layer is append-only. No transforms. No business logic. Just get the data in and preserve it exactly as it arrived so you can always replay from source. Python from pyspark.sql import SparkSession from pyspark.sql.functions import current_timestamp, lit from delta.tables import DeltaTable spark = SparkSession.builder.getOrCreate() # Read raw events from ADLS Gen2 / Event Hub / source of choice raw_events = spark.read.format('json').load('abfss://[email protected]/events/') # Add ingestion metadata — never mutate source columns bronze_df = raw_events.withColumn('_ingested_at', current_timestamp()) \ .withColumn('_source', lit('events_api')) # Write to Bronze Delta table — append only, no overwrites bronze_df.write \ .format('delta') \ .mode('append') \ .option('mergeSchema', 'true') \ .saveAsTable('churn.bronze.events') print(f"Bronze rows written: {bronze_df.count()}") Why append-only? If your downstream pipeline produces bad features, you want to replay from Bronze without re-ingesting from source. Overwriting Bronze breaks that ability. Step 2 — Silver Layer: Clean and Validate Silver is where you enforce schema, handle nulls, deduplicate, and standardize. Think of it as your canonical, trusted dataset. Python from pyspark.sql.functions import col, to_timestamp, when, trim, upper from delta.tables import DeltaTable bronze = spark.table('churn.bronze.events') silver_df = bronze \ .filter(col('customer_id').isNotNull()) \ .filter(col('event_type').isNotNull()) \ .dropDuplicates(['customer_id', 'event_id']) \ .withColumn('event_ts', to_timestamp(col('event_timestamp'))) \ .withColumn('event_type', upper(trim(col('event_type')))) \ .withColumn('country_code', when(col('country').isNull(), lit('UNKNOWN')) .otherwise(upper(col('country')))) \ .select( 'customer_id', 'event_id', 'event_type', 'event_ts', 'country_code', 'product_id', 'session_id', '_ingested_at', ) # Upsert into Silver using Delta MERGE — idempotent on re-runs if DeltaTable.isDeltaTable(spark, 'churn.silver.customers'): silver_table = DeltaTable.forName(spark, 'churn.silver.customers') silver_table.alias('tgt').merge( silver_df.alias('src'), 'tgt.customer_id = src.customer_id AND tgt.event_id = src.event_id' ).whenNotMatchedInsertAll().execute() else: silver_df.write.format('delta').saveAsTable('churn.silver.customers') print(f"Silver table updated. Total rows: {spark.table('churn.silver.customers').count()}") Step 3 — Gold Layer: Feature Engineering This is the heart of the pipeline. We compute aggregated, windowed, and encoded features that the model will actually train on. Python from pyspark.sql.functions import ( col, count, countDistinct, sum as _sum, avg, datediff, max as _max, min as _min, current_date, expr, when ) from pyspark.sql.window import Window silver = spark.table('churn.silver.customers') # ------------------------------------------------------------------ # 1. Aggregate features per customer over 30 / 90 day windows # ------------------------------------------------------------------ today = current_date() agg_features = silver \ .withColumn('days_since_event', datediff(today, col('event_ts'))) \ .groupBy('customer_id') \ .agg( count('event_id') .alias('total_events'), countDistinct('session_id') .alias('total_sessions'), countDistinct('product_id') .alias('distinct_products'), _sum(when(col('days_since_event') <= 30, 1).otherwise(0)) .alias('events_last_30d'), _sum(when(col('days_since_event') <= 90, 1).otherwise(0)) .alias('events_last_90d'), _max('event_ts') .alias('last_event_ts'), _min('event_ts') .alias('first_event_ts'), ) \ .withColumn('days_since_last_event', datediff(today, col('last_event_ts'))) \ .withColumn('customer_tenure_days', datediff(today, col('first_event_ts'))) \ .withColumn('avg_events_per_day', col('total_events') / (col('customer_tenure_days') + 1)) # ------------------------------------------------------------------ # 2. Encode churn risk tier as ordinal feature # ------------------------------------------------------------------ feature_df = agg_features \ .withColumn('recency_tier', when(col('days_since_last_event') <= 7, lit(3)) # active .when(col('days_since_last_event') <= 30, lit(2)) # at risk .otherwise(lit(1)) # churned ) \ .withColumn('engagement_score', (col('events_last_30d') * 0.6 + col('events_last_90d') * 0.4) / (col('customer_tenure_days') + 1) ) # ------------------------------------------------------------------ # 3. Write to Gold feature store — overwrite with partition by date # ------------------------------------------------------------------ feature_df \ .withColumn('feature_date', current_date()) \ .write \ .format('delta') \ .mode('overwrite') \ .option('replaceWhere', f"feature_date = '{today}'") \ .saveAsTable('churn.gold.features') print(f"Gold features written: {feature_df.count()} customers") Step 4 — MLflow: Track the Training Run With features in Gold, we hand off to MLflow to train, track, and register the model. Notice we log the Delta table version so we can always reproduce exactly which feature snapshot trained which model. Python import mlflow import mlflow.sklearn from mlflow.models.signature import infer_signature from sklearn.ensemble import GradientBoostingClassifier from sklearn.model_selection import train_test_split from sklearn.metrics import roc_auc_score, f1_score import pandas as pd mlflow.set_experiment('/churn-prediction/feature-pipeline') # Read Gold features — capture Delta version for reproducibility gold_table = DeltaTable.forName(spark, 'churn.gold.features') delta_version = gold_table.history(1).select('version').collect()[0][0] features_pdf = spark.table('churn.gold.features').toPandas() FEATURE_COLS = [ 'total_events', 'total_sessions', 'distinct_products', 'events_last_30d', 'events_last_90d', 'days_since_last_event', 'customer_tenure_days', 'avg_events_per_day', 'recency_tier', 'engagement_score', ] TARGET = 'churned' X = features_pdf[FEATURE_COLS] y = features_pdf[TARGET] X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42) with mlflow.start_run(run_name=f'gbm-features-v{delta_version}') as run: params = {'n_estimators': 200, 'max_depth': 5, 'learning_rate': 0.05} model = GradientBoostingClassifier(**params, random_state=42) model.fit(X_train, y_train) y_pred = model.predict(X_test) y_prob = model.predict_proba(X_test)[:, 1] # Log everything mlflow.log_params(params) mlflow.log_metric('roc_auc', roc_auc_score(y_test, y_prob)) mlflow.log_metric('f1_score', f1_score(y_test, y_pred)) mlflow.log_param('delta_feature_version', delta_version) mlflow.log_param('feature_columns', FEATURE_COLS) mlflow.log_param('training_rows', len(X_train)) # Log model with signature signature = infer_signature(X_train, y_pred) mlflow.sklearn.log_model( model, artifact_path='churn-gbm', signature=signature, registered_model_name='churn-prediction-gbm', ) print(f"Run ID: {run.info.run_id}") print(f"ROC-AUC: {roc_auc_score(y_test, y_prob):.4f}") print(f"Feature Delta version logged: {delta_version}") Bonus: Delta Lake Time Travel for Feature Reproducibility One of the best things about Delta Lake is time travel. If a model behaves unexpectedly in production, you can reload the exact feature snapshot it was trained on. Python # Reload the exact feature version that trained a specific model run import mlflow run = mlflow.get_run('your-run-id-here') feature_version = int(run.data.params['delta_feature_version']) # Rehydrate that exact feature snapshot historical_features = spark.read \ .format('delta') \ .option('versionAsOf', feature_version) \ .table('churn.gold.features') print(f"Loaded feature snapshot from Delta version {feature_version}") print(f"Row count: {historical_features.count()}") # You can now retrain on the exact same data to reproduce the result Service Comparison ToolRole in pipelineWhy not the alternativeApache SparkDistributed feature computationPandas (single node, OOM at scale), Dask (less native Databricks integration)Delta LakeFeature storage with versioningParquet (no ACID, no time travel), Hive tables (no merge support)MLflow TrackingExperiment and param loggingManual logging (not reproducible), W&B (extra cost, less native on Databricks)MLflow RegistryModel versioning and promotionCustom model store (more ops overhead)Medallion ArchitecturePipeline layer separationFlat pipelines (hard to debug, no replay capability)Delta MERGEIdempotent Silver upsertsOverwrite (destroys history), append (creates duplicates) Things to Watch in Production Shuffle partitions matter. Spark defaults to 200 shuffle partitions, which is fine for small data but will bottleneck at scale. Set spark.conf.set("spark.sql.shuffle.partitions", "auto") on Databricks Runtime 10+ or tune it manually to 2-3x your core count. Z-ordering on Gold features. If you're querying Gold by customer_id frequently, add OPTIMIZE churn.gold.features ZORDER BY (customer_id) after the write. This co-locates related data and cuts query times dramatically on large tables. Log Delta version in every MLflow run. This is non-negotiable for reproducibility. Without it you can't prove which feature snapshot trained which model, which becomes a compliance problem in regulated industries. Cluster autoscaling for feature jobs. Feature engineering jobs tend to have spiky resource needs (big during aggregation, small during writes). Enable autoscaling on your Databricks cluster and set a min/max node count rather than a fixed size. Wrapping Up The combination of Spark, Delta Lake, and MLflow on Databricks gives you a feature engineering pipeline that is reproducible (Delta time travel + MLflow param logging), scalable (Spark handles billions of rows), and auditable (every run is tracked, every feature version is stored). The Medallion Architecture keeps the pipeline modular — you can rerun just the Gold layer if you change a feature definition without touching Bronze or Silver, and MLflow ties model performance back to the exact feature version that produced it. References Azure Databricks DocumentationDelta Lake — The Definitive GuideApache Spark SQL — Window FunctionsMLflow Tracking DocumentationMLflow Model RegistryMedallion Architecture on DatabricksDelta Lake Time TravelDatabricks Feature Store Overview

By Jubin Abhishek Soni CORE

The work that happens while your app is not in the foreground has always been the fiddly part of mobile development, and Codename One's coverage of it had gaps. PR #5142 modernizes local notifications, push, background execution, and shared content across the core, JavaSE, Android, and iOS, and importantly, it makes all of it work in the simulator so you can iterate without a device. Background Work With Constraints The new com.codename1.background package schedules work that the OS runs when its conditions are met, mapping to Android JobScheduler and iOS BGTaskScheduler underneath. You describe what the work needs, not when to poll: Java WorkRequest req = WorkRequest.builder("daily-sync", SyncWorker.class) .setRequiresNetwork(true) .setRequiresCharging(true) .setPeriodic(6 * 60 * 60 * 1000L) .build(); BackgroundWork.schedule(req); The worker is a small class with a no-argument constructor that the platform instantiates when it runs your task: Java public class SyncWorker implements BackgroundWorker { public void performWork(String workId, Map<String, String> inputData, long deadline, Callback<Boolean> onComplete) { boolean ok = pullLatestData(); onComplete.onSucess(ok); } } The builder covers the usual constraint set: network, unmetered network, charging, idle, battery-not-low, periodic intervals, an initial delay, and input data. For longer foreground operations, there is ForegroundService.start(...), which runs a JVM task behind a persistent notification on Android, and for heavier iOS background processing BackgroundTask.scheduleProcessing(...) maps to BGProcessingTaskRequest. The iOS background-processing identifiers are declared with the ios.backgroundProcessingIds build hint, which the builder turns into the matching Info.plist entries for you. Notifications Got a Lot Richer LocalNotification gained a long list of capabilities, all added backward-compatibly so every existing field, getter, and setter behaves exactly as before. New on top of that: an image attachment, multiple action buttons, inline quick reply, per-channel sound, grouping with a summary, full-screen intent, time-sensitive delivery, ongoing and progress notifications, a custom view, and a messaging-conversation style. A download-progress notification, for example: Java LocalNotification n = new LocalNotification(); n.setId("download"); n.setAlertTitle("Downloading"); n.setAlertBody("episode-12.mp4"); n.setOngoing(true); n.setProgress(100, 40); Display.getInstance().scheduleLocalNotification( n, System.currentTimeMillis(), LocalNotification.REPEAT_NONE); Or a notification with an inline reply, the kind a messaging app uses: Java n.addInputAction("reply", "Reply", "Type a message", "Send"); On newer Android devices, notification channels are now first-class through a builder routed via Display: Java Display.getInstance().registerNotificationChannel( new NotificationChannelBuilder("messages", "Messages") .importance(NotificationChannelBuilder.IMPORTANCE_HIGH) .enableVibration(true) .lockscreenVisibility(NotificationChannelBuilder.VISIBILITY_PRIVATE)); Permission requests are explicit too, with Display.requestNotificationPermission(...) taking a NotificationPermissionRequest (provisional, critical, time-sensitive, or the Android POST_NOTIFICATIONS permission) and returning a NotificationPermissionResult you can check with isGranted(). Push Topics Push now supports topic subscriptions: Java Push.subscribeToTopic("sports"); Push.unsubscribeFromTopic("sports"); On Android these map to FCM topics. On iOS they are a documented no-op, because raw APNs have no topic concept; the call is safe to make on both, so your code stays cross-platform. Receiving Shared Content If a user shares text, a URL, a file, or an image into your app from another app, it now arrives through a single lifecycle hook: Java public void onReceivedSharedContent(SharedContent content) { // content carries text / url / file / image items } On Android this is backed by a share-receiver activity that handles SEND and SEND_MULTIPLE and hands files off through app storage. On iOS it reuses the share extension that landed two weeks ago and reads the App-Group payload when the app activates; the App Group is configured with the ios.shareAppGroup build hint. The build plugin wires the manifest entries, services, and intent filters automatically based on a classpath scan, so turning these features on does not mean hand-editing platform descriptors. All of It Runs in the Simulator The piece that makes this practical day-to-day is the JavaSE support. There is a new "Notifications and Background" entry in the Simulate menu with constraint toggles, a run-the-work-now button, a channel inspector, and shared-content injection, plus a rich notification panel that renders images, actions, inline quick reply, and progress and routes taps back to your LocalNotificationCallback and PushContent on the same code path the device uses. You can build and debug these flows entirely on your desktop before you ever make a build. A control screen like this, with the scheduled job and the actions wired to the calls above, runs and renders in the simulator: The previous deep dive was about the new advertising API, and the release post has the full index for the week, including the smaller fixes and the note about how we are handling contributions now. Keep an eye out for our next release this Friday.

By Shai Almog CORE

Every business runs on a database, but not everyone who needs an answer from the database speaks SQL. Data Analysts wait on engineers, and stakeholders wait on analysts, and by the time the query runs, the decision window has passed. LangChain's SQL integration fixes this, translating plain English questions like "Which product category had the highest revenue last year' into valid SQL, executing it, and returning a human-readable answer. In this tutorial, we will build a full natural-language SQL interface that covers setup, complex queries, and safety guardrails. How It Works LangChain's SQL integration follows a three-step pattern: Question -> SQL generation -> execution -> Natural Language Answer The LLM only sees the schema and never the RAW data. Step 1: Setup Shell pip install langchain langchain-openai langchain-community\ sqlalchemy pymysql psycopg2-binary Shell import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() llm = ChatOpenAI( model="gpt-4o" temperature = 0 #keep deterministic for SQL generation ) Step 2: Connecting to the Database LangChains' SQLDatabase wrapper works with any SQLAlchemy-compatible database. SQLite(Local/Dev): Python from langchain_community.utilities import SQLDatabase #Connect to local SQLite database db = SQLDatabase.from_uri("sqlite:///sales.db") #Check what LangChain can see print(db.get_usable_table_names()) #['customers','orders','products','order_items'] print(db.get_table_info()) # Create Table Customers ( # id INTEGER PK # name varchar() # email varchar() # region varchar() # created_time For PostgreSQL: Python db = SQLDatabase.from_uri("postgresql+psycopg2:/user:password@localhost:5432/test_db") For MySQL: Python db = SQLDatabase.from_uri("mysql+pymysql://user:password@localhost:3306/test_db") Large databases: Use include_tables to limit schema exposure and sample_rows_in_table_info to show LLM real data formats: Python db = SQLDatabase.from_uri( "postgresql+psycopg2://user:password@localhost/test_db") include_tables=["customers","orders","products","order_items"], sample_rows_in_table_info=2 ) Step 3: Sample Schema We will use a simple e-commerce schema: customers, products, orders, and order_items: SQL CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT,email TEXT, region TEXT,created_at DATETIME); CREATE TABLE products (id INTEGER PRIMARY KEY, name TEXT,category TEXT, price INTEGER,stock INTEGER); CREATE TABLE orders (id INTEGER PRIMARY KEY, cust_id INTEGER,email TEXT, STATUS TEXT,created_at DATETIME); CREATE TABLE order_items (id INTEGER PRIMARY KEY, ORDER_ID INTEGER,email TEXT, PRODUCT TEXT,QUANTITY INTEGER, UNIT_PRICE INTEGER); Step 4: Basic Natural Language Queries LangChain provides create_SQL_query_chain for the SQL generation step: Python for LangChain.chains import sql_query_chain #create a chain that coverts question to SQL sql_chain = create_sql_query_chain(llm,db) #Generate SQL from natural language question question = "How many customer do we have in each region" sql_query = sql_chain.invoke({"question":question}) print(f"Generated SQL:\n{sql_query}") #Select region,COUNT(*) as customer_count #From customers #Group By region #order by customer_count desc To get the results for this query, add the query executor: Python from langchain_community.tools import QuerySQLDataBaseTool #Tool that executes SQL against the database execute_query = QuerySQLDataBaseTool(db=db) #Chain: question>SQL->execute->raw_result result= execute_query.invoke(sql_query) print(result) #[('North',14),('South',13),('East',12),('West',11) To get natural language rather than tuples, chain it with LCEL: Python from langchain_core.output_parsers import StrOuputParser from langchain_core.prompts import PromptTemplate from operator import itemgetter #Prompt to convert SQL results into Human readable answers answer_prompt = PromptTemplate.from_template( """Given the following user question,SQL query and SQL result Write a clear and concise answer in natural language. Question:{question} SQL Query:{query} SQL Result:{result} Answer:""" ) #Full pipeline: Question -> SQL-> execute -> Natural Language full_chain = ( {"question":itemgetter("question"), "query":sql_chain} |{"question": itemgetter("question"), "query": itemgetter("query"), "result": lambda x: execute_query.invoke(x["query"])} | answer_prompt | llm | StrOutputParser() ) response = full_chain.invoke({"question":"How many customers do we have in each region"}) LCEL: LangChain Expression Language It's LangChain's declarative syntax for chaining components together using the pipe(|) operator borrowed from Unix style. Instead of manually writing inputs and outputs between steps, it is composed from right to left. Step 5: Handling Complex Multi-Table Queries The real test of SQL chain is multi-table queries involving joins, aggregations, and filters. Python questions = [ "What are the top 3 best selling categories by total revenue this month", "Which customers have placed more than 5 orders", "What is the average order value by region?", ] for q in questions: answer = full_chain.invoke({"question":q}) print(f"Q:{q}\nA:{answer}\n") Sample output: Q: What are the top 3 best-selling categories by total revenue this month? A: Electronics leads at $8,432.50, followed by Sports ($2,109.75) and Clothing ($1,876.20). Step 6: Handling Complex Multi-Table Queries Giving an LLM access to your database requires careful guardrails. A poorly crafted question, or a malicious one, could result in DELETE,DROP, or UPDATE statements. Approach 1: Read-Only Database Connection The simplest safeguard: connect with a read-only user. Python #SQL Read Only User db = SQLDatabase.from_uri("postgresql+psycopg2://readonly_user:password@localhost/test_db") Approach 2: Query Validation Before Execution Python import re FORBIDDEN_KEYWORDS = ["DROP","DELETE","INSERT","UPDATE","ALTER","GRANT","REVOKE"] def validate_sql(query: str)-> tuple[bool,str]: """Check SQL Query for destructive operations.""" query_upper = query.upper().strip() for keyword in FORBIDDEN_KEYWORDS: #MATCH WHOLE WORDS ONLY AVOID FALSE POSITIVES LIKE UPDATES IN COLUMN NAME if re.search(rf"\b{keyword}\b", query_upper): return False, f"Blocked, query contains forbidden keyword '{keyword}'" if not query_upper.startswith("SELECT"): return False, "Blocked:Only select queries are permitted" return True, "OK" def safe_execute(query:str) -> str: """Validate and then Execute a query""" is_safe,reason = validate_sql(query) if not is_safe: return f"query_rejected:{reason}" return execute_query.invoke(query) Approach 3: Custom System Prompt Python from langchain_core.prompts import ChatPromptTemplate safe_sql_prompt = ChatPromptTemplate.from_messages([ ("system","""You are an SQL expert generate only SELECT queries CRITICAL RULES: - NEVER Generate INSERT,DELETE,UPDATE,ALTER,DROP OR TRUNCATE QUERIES - NEVER USE SEMI COLON TO CHAIN STATEMENTS - Only query table listed in the Schema - Limit result to 1000 unless specified Database dialect: {dialect} Available tables and schema: {table_info}""", ("human","{input}") ]) Step 7: Building Conversational SQL Agent In production, you will need an agent that can handle follow-up questions, correct its own mistakes, and retry failed queries. LangChain's agent framework handles all of this. Python from langchain_community.agents_toolkits import create_sql_agent from langchain_community.agent_toolkits.sql.toolkit import SQLDatabaseToolkit #The toolkit bundles all SQL related tools #- sql_db_query: Execute SQL #- sql_db_schema: Get Table Name #- sql_db_list_table: List available Tables #- sql_db_query_checker: Validate SQL before running toolkit = SQLDatabaseToolkit(db=db,llm=llm) agent = create_sql_agent( llm=llm, toolkit = toolkit, verbose = True, #show reasoning step agent_type = "openai-tools", max_iterations =10, handle_parsing_errors=True #prevent infinite loop ) #Agent can handle multi-step reasoning response = agent.invoke({ "input":"Find the customer who spent the most money in last 30 days," "and tell me the products they bought." print(response["output"]) Step 8: Add Memory and CLI Interface Add ConversationBufferWindowMemory so that the user can ask follow-up questions, then wrap everything in the interactive loop. Python from langChain.memory import ConversationBufferWindowMemory memory = ConversationBufferWindowMemory( memory_key="chat_history", return_messages=True, k=5 ) agent_with_memory = create_sql_agent( llm=llm, toolkit = toolkit, agent_type="openai-tools", memory=memory, handle_parsing_errors= TRUE ) #follow up question resolve context automatically agent_with_memory.invoke({"input":"what was your total revenue last month"}) agent_with_memory.invoke({"input":"which region contributed the most"}) agent_with_memory.invoke({"input":"who were the top customers there"}) #CLI loop while True: q=input("You: ").strip() if q.lower() =="quit":break print("Agent:",agent_with_memory.invoke({"input": q})["output"]) Common Pitfalls Ambiguous column names across tables: Tell the LLM which particular table to refer to or use include_ tables parameter to limit scope.Hallucinated table or column name: Always pass sample_rows_in_table=2 so the model sees real data format and is less likely to invent column names.Slow queries on large tables: Add a LIMIT instruction to your prompt: " Always add limit 100 unless the user asks for more."Date/Time dialect difference: SQLite uses date('now','-30 days') while PostgreSQL uses NOW() -interval '30 days'. LangChain passes the dialect to the LLM; hence, clearly specify it in the prompt.Token limit on wide schemas: For a database with 50+ tables, use include_tables to pass only a relevant subset per query, so specify it clearly in your prompt Conclusion LangChain's SQL integration removes the translation layer between human and data. Business users get instant answers, analysts focus on interpretation rather than query writing, and the underlying database remains unchanged. The post covers: natural_language->SQL->answer pipeline, multi-table joins, three layers of safety guardrails, self-correcting agent, and conversation memory. You can use this with BigQuery or Snowflake as well.

Industrial control systems are generating more data than ever before, but the Python tooling used to process this telemetry often encounters severe performance constraints. Traditional OPC UA libraries are built around synchronous, polling-based Client and Server architectures. When industrial networks scale to thousands of sensors broadcasting high-frequency data, these synchronous Python implementations choke. To handle this modern many-to-many topology, developers need a native Publisher and Subscriber solution that does not block the execution thread while waiting for network packets. For Python developers unfamiliar with industrial protocols, OPC UA PubSub (IEC 62541-14) is a standard that decouples data producers from consumers by allowing devices to broadcast telemetry via stateless middleware like UDP Multicast. For industrial engineers new to Python concurrency, asyncio is a standard library that uses an event loop to handle thousands of simultaneous network operations concurrently without the heavy overhead of traditional threading. Bridging these two paradigms requires a completely non-blocking architecture. To address this gap, a complete asyncio driven OPC UA PubSub implementation was architected and integrated into the open source opcua-asyncio library (merged in Commit 2b6f3e5). Implementing this standard from scratch in an asynchronous Python environment presented unique challenges. This article breaks down the engineering decisions and technical design patterns used to build this extension. By contributing this capability to a library that serves thousands of developers in the Python IIoT ecosystem, the goal is to ensure engineers can now build highly scalable publisher and subscriber sensor networks without migrating away from Python. The Shift to Publisher and Subscriber in IIoT In traditional OPC UA, a client polls a server or sets up monitored items. This creates a tightly coupled, connection-oriented topology. The PubSub extension decouples this by allowing publishers to broadcast telemetry data via stateless middleware like UDP Multicast or MQTT, which subscribers can passively ingest. To bring this to the opcua-asyncio ecosystem, the architecture needed to bridge the gap between Python's asynchronous event loop and the highly deterministic, byte-packed UADP (OPC UA Datagram Protocol) structures. The design was broken down into four core pillars. Asynchronous transport layer: Managing non-blocking UDP and IP multicast.UADP binary protocol engine: Bit-level packing and unpacking of network messages.Data abstraction and node mapping: Linking arbitrary network payloads to the OPC UA Address Space.Concurrency and connection management: Orchestrating readers, writers, and tasks via asyncio. Pillar 1: The Asynchronous UDP Transport Layer OPC UA UADP relies on UDP for low-latency transmission. In Python, synchronous socket operations block the main thread, which is fatal to an asyncio application. To solve this, the networking layer was built directly on top of asyncio.DatagramProtocol. The OpcUdp class overrides the standard protocol callbacks to bridge the network socket with the PubSub receiver logic. Here is a look at how the protocol was extended and hooked into the event loop to ensure incoming datagrams never block the main thread. Python class OpcUdp(asyncio.DatagramProtocol): def __init__(self, cfg: UdpSettings, receiver: Optional[PubSubReceiver], publisher_id: Variant) -> None: super().__init__() self.cfg = cfg self.receiver = receiver self.publisher_id = publisher_id.Value def datagram_received(self, data: bytes, source: Tuple[str, int]) -> None: try: buffer = Buffer(data) msg = UadpNetworkMessage.from_binary(buffer) if self.receiver is not None: asyncio.ensure_future(self.receiver.got_uadp(msg)) except Exception: logging.exception("Received Invalid UadpPacket") Socket lifecycle: The UdpSettings class manages socket creation by carefully applying SO_REUSEADDR and handling both IPv4 (AF_INET) and IPv6 (AF_INET6) multicasting.Multicast configuration: Depending on the IP family, IP_ADD_MEMBERSHIP or IPV6_JOIN_GROUP are injected directly into the socket options via the struct module to ensure the application correctly subscribes to IGMP or MLD groups.Non-blocking reception: When a datagram hits the interface, datagram_received immediately passes the raw bytes to the UADP decoding engine and dispatches the resulting parsed message to a background task using asyncio.ensure_future(). This guarantees the networking thread is instantly freed to handle the next packet. Pillar 2: The UADP Binary Protocol Engine The UADP specification defines an extremely dense, highly variable network packet. Headers can dynamically expand or contract based on a series of bit flags. Processing this in Python requires rigorous byte manipulation to maintain both memory efficiency and processing speed. The uadp.py implementation utilizes Python's enum.IntFlag to map the exact bitwise schemas defined in OPC UA Part 14. Python class MessageHeaderFlags(IntFlag): NONE = 0 UADP_VERSION_BIT0 = 0b1 PUBLISHER_ID = 0b00010000 GROUP_HEADER = 0b00100000 PAYLOAD_HEADER = 0b01000000 EXTENDED_FLAGS_1 = 0b10000000 # FlagsExtend1 PUBLISHER_ID_UINT16 = 0b0000000100000000 PUBLISHER_ID_UINT32 = 0b0000001000000000 PUBLISHER_ID_UINT64 = 0b0000011000000000 PUBLISHER_ID_STRING = 0b0000010000000000 Flag-driven serialization: The UadpHeader and UadpDataSetMessageHeader are deeply nested and conditional. For example, the Extended Flags dictate whether a PublisherId is encoded as a Byte, UInt16, UInt32, UInt64, or String.Bitwise extensibility: The implementation cascades flags using EXTENDED_FLAGS_1 and EXTENDED_FLAGS_2 bits. If the integer value of the required flags exceeds 0xFF, the engine dynamically shifts the bytes and appends the extension flags.Binary packing: A standardized Primitives unpacking utility translates the raw buffer directly into strictly typed Python objects like UInt32, Guid, or DateTime. This avoids the overhead of intermediate object instantiation when parsing high-frequency sensor data.Delta Frames vs. raw data: The parser dynamically routes payload deserialization based on MessageDataSetFlags. It distinguishes between Key Frames, Delta Frames, and Raw Data while packing the resulting generic DataValue structs into a unified UadpNetworkMessage. Pillar 3: Data Abstraction and Address Space Integration Receiving data is only half the battle because that data must meaningfully map to the server's Address Space. The architecture introduces PubSubInformationModel to handle this synchronization. Datasets and metadata: A PublishedDataSet defines the structure of the data being transmitted. This includes tracking FieldMetaData, built in types, and value ranks.Dynamic variable substitution: The PubSubDataSourceServer class abstracts the retrieval of data from the server tree. It safely reads attributes and falls back to a SubstituteValue if a node status code is bad. This ensures unbroken telemetric streams.Subscribed mirrors: When an OPC UA client acts as a subscriber, it needs to see the incoming data reflected in its own node tree. The SubscribedDataSetMirror dynamically creates new variable nodes on the fly to match the incoming DataSetMetaData. This dynamic node mapping was engineered by injecting new variables straight into the server tree based on the metadata specification. Python async def _create_and_set_node(self, f: FieldMetaData): if self._node is None: raise RuntimeError("SubscribedDataSetMirror._node is not initialized.") n = await self._node.add_variable( NodeId(NamespaceIndex=Int16(1)), "1:" + str(f.Name), Variant(), datatype=f.DataType ) await n.write_attribute(AttributeIds.Description, f.Description) await n.write_attribute(AttributeIds.ValueRank, f.ValueRank) await n.write_attribute(AttributeIds.ArrayDimensions, f.ArrayDimensions) return n Target variables: Alternatively, SubScribedTargetVariables maps incoming dataset fields directly to existing NodeId references in the server. These references update in real time as UDP packets are decoded. Pillar 4: Concurrency and Connection Management The top-level orchestration is handled by the PubSubConnection and PubSub classes. These act as the asynchronous lifecycle managers. Task gathering: When start() is invoked on a connection, the lifecycle manager utilizes asyncio.gather() to concurrently spin up all associated DataSetReader and DataSetWriter tasks without blocking the main OPC UA server loop. Python async def start(self) -> None: logging.info("Starting Connection %s", await self.get_name()) loop = asyncio.get_event_loop() sock, _, _ = self._network_settings.create_socket() self._transport, self._protocol = await loop.create_datagram_endpoint( lambda: self._network_factory(self._network_settings, self._receiver, self._cfg.PublisherId), sock=sock, ) self._writer_tasks = asyncio.gather(*[writer.run(self._protocol, self._app) for writer in self._writer_groups]) reader_tasks = asyncio.gather(*[reader.start() for reader in self._reader_groups]) await reader_tasks if self._protocol is not None: self._protocol.set_receiver(self._receiver) await self._set_state(PubSubState.Operational) Protocol decoupling: To prevent circular dependencies between the network transport and the information model, strict interfaces defined in protocols.py are used. The UDP protocol layer communicates with the logical layer strictly through these abstract protocols.Wildcard routing and readers: The ReaderGroup acts as an intelligent multiplexer. When a multi-payload UADP packet arrives, it analyzes the GroupHeader and DataSetPayloadHeader. It then routes individual DataSetMessages to the correct DataSetReader instances by matching wildcard filters.Timeouts and state machines: Robust industrial systems must handle connection drops. The DataSetReader wraps its operation in a dedicated timeout task. Using asyncio.wait_for(), it actively monitors for MessageReceiveTimeout events. If a heartbeat or payload is missed, it transitions the internal PubSubState to Error. This allows higher-level application logic to gracefully degrade. Conclusion Building a production-ready OPC UA PubSub stack in Python requires harmonizing the stringent bit-packed demands of the IEC 62541-14 specification with the asynchronous paradigms of asyncio. By leveraging asyncio.DatagramProtocol for deterministic networking, abstracting the UADP bit flags into structured classes, and deeply integrating with the OPC UA Address space via mirrored target variables, this implementation provides a scalable foundation for modern IIoT architectures. Code and Open Source Contributions The architecture and implementation details discussed in this article were merged into the core FreeOpcUa/opcua-asyncio repository. You can explore the complete implementation, including the raw protocol parsing and asyncio abstractions, via the links below. Primary commit: 2b6f3e5 (Initial implementation of OPC UA PubSub UDP and UADP). Key files to explore in the commit: asyncua/pubsub/udp.py: Contains the OpcUdp transport layer and multicast socket configuration.asyncua/pubsub/uadp.py: Houses the flag driven serialization and binary protocol engine.asyncua/pubsub/connection.py: Demonstrates the asyncio task management and lifecycle orchestration.

Modern enterprise applications rarely operate in isolation. A user may authenticate through a web or mobile application, invoke a Java-based backend API, and that backend may need to call additional downstream services such as microservices or third-party APIs. In these scenarios, simply using the application's identity is often insufficient. The downstream service may need to know which user initiated the request and enforce authorization based on that user's permissions. This is where the OAuth 2.0 On-Behalf-Of (OBO) flow becomes invaluable. In this article, I will summarize how the OBO flow works, where it fits in a modern Java architecture, and how to implement it securely in a Spring Boot application. How Does Each Downstream Service Know Who the Original User Is? One of the first assumptions many engineers make is that the backend can simply reuse its own application credentials when communicating with another service. While this works for machine-to-machine communication, it falls short whenever user-specific authorization is required. Consider a healthcare application where a physician logs into a patient portal and requests medical records. The initial Java API authenticates the request, but retrieving those records may require calling another internal API responsible for patient information. That downstream API needs to know which physician initiated the request before deciding whether access should be granted. If the Java backend uses only its own application identity, the downstream service loses the user context and cannot perform authorization based on the physician's permissions. This is exactly the problem that the OAuth 2.0 On-Behalf-Of (OBO) flow was designed to solve. What Is OBO (On-Behalf-Of) Flow? The OBO flow allows a middle-tier service (API A) to obtain an access token for another downstream service (API B) while preserving the identity and permissions of the signed-in user. Instead of API A calling API B using its own application credentials, API A exchanges the user's access token for a new token intended for API B. The flow looks like this: Plain Text User | v Web/Mobile Applications | | Access Token v Java API A | | OBO Token Exchange v Identity Provider | | New Access Token v Java API B As a result, API B receives a token representing the actual user, allowing it to perform proper authorization checks. Why Not Use Client Credentials? Many developers mistakenly use the Client Credentials flow when calling downstream APIs. While Client Credentials works for service-to-service communication, it does not carry user context. Consider a healthcare application like ours: Dr. Smith logs into a patient portal.The Java API retrieves patient records from another service.The downstream service must verify Dr. Smith's permissions. If Client Credentials is used, the downstream service only sees the application identity and loses visibility into the actual user making the request. OBO solves this problem by preserving delegated permissions. Typical Enterprise Use Cases OBO is commonly used for Healthcare applications accessing patient records, Enterprise microservices, Multi-tier API architectures, internal service authorization, and audit and compliance requirements. Many organizations implementing zero-trust architectures rely heavily on delegated authorization models such as OBO. Implementing OBO in Spring Boot Let's assume the following: Microsoft Entra ID (Azure AD) is the Identity Provider.API A is a Spring Boot application.API B is a downstream service. Step 1: Add MSAL4J Dependency XML <dependency> <groupId>com.microsoft.azure</groupId> <artifactId>msal4j</artifactId> <version>1.15.0</version> </dependency> Step 2: Acquire Token On Behalf Of User The incoming access token is received from the frontend application. Java String clientId = "YOUR_CLIENT_ID"; String clientSecret = "YOUR_CLIENT_SECRET"; IClientCredential credential = ClientCredentialFactory.createFromSecret(clientSecret); ConfidentialClientApplication app = ConfidentialClientApplication.builder( clientId, credential) .authority( "https://login.microsoftonline.com/TENANT_ID") .build(); UserAssertion userAssertion = new UserAssertion(incomingUserToken); OnBehalfOfParameters parameters = OnBehalfOfParameters.builder( Collections.singleton("api://api-b/.default"), userAssertion) .build(); IAuthenticationResult result = app.acquireToken(parameters).join(); String downstreamAccessToken = result.accessToken(); At this point, the Java application has obtained a new token that can be used to call API B while preserving the user's identity. Step 3: Call the Downstream API Using Spring's RestTemplate: Java HttpHeaders headers = new HttpHeaders(); headers.setBearerAuth(downstreamAccessToken); HttpEntity<Void> request = new HttpEntity<>(headers); ResponseEntity<String> response = restTemplate.exchange( "https://api-b.company.com/patients", HttpMethod.GET, request, String.class); return response.getBody(); API B now receives a delegated token representing the authenticated user. Security Best Practices Implementing OBO correctly is critical. 1. Validate Incoming Tokens Always validate: SignatureIssuerAudienceExpiration Never trust tokens received from clients without validation. 2. Apply Least Privilege Only request scopes required by the downstream API. Bad: Plain Text https://graph.microsoft.com/.default Better: Plain Text User.Read Limiting scopes reduces the blast radius if a token is compromised. 3. Never Log Access Tokens Avoid: Plain Text logger.info(token); Access tokens often contain sensitive claims and permissions. 4. Secure Client Secrets Store secrets in: Azure Key VaultAWS Secrets ManagerHashiCorp Vault Avoid storing secrets in: Plain Text application.properties or source code repositories. 5. Implement Token Caching Repeated token acquisition creates unnecessary latency. Consider caching OBO tokens until they expire. Most enterprise identity libraries already provide token caching support. Common Mistakes Some common issues I frequently encounter when onboarding new developers include: Using Client Credentials instead of OBOPassing user tokens directly to downstream APIsRequesting excessive scopesLogging JWT tokensNot validating token audiencesHardcoding client secrets These mistakes often lead to authorization failures or security vulnerabilities. Conclusion As organizations adopt microservices and API-first architectures, preserving user identity across service boundaries becomes increasingly important. The OAuth 2.0 On-Behalf-Of flow provides a secure and standards-based approach for allowing Java applications to call downstream APIs while maintaining the original user's context and permissions. By implementing OBO correctly, developers can build applications that are more secure, auditable, and aligned with modern zero-trust security principles. For enterprise Java teams, understanding OBO is no longer optional; it is becoming a fundamental requirement for building secure distributed systems.

Most recommendation systems are batch jobs. They crunch last night's data, write a recommendations table, and serve it all day. That works fine until your user watches three thriller movies in a row at 9 pm and your system is still recommending rom-coms because the batch hasn't run yet. In this post, I'll walk through building an agent system that reacts to streaming user behavior in real time using: Amazon Kinesis to ingest and route eventsAWS Lambda to process, enrich, and trigger reasoningAmazon Bedrock as the reasoning and recommendation layerDynamoDB to store user profiles and recommendation cacheS3 for raw event archiving and model artifacts By the end, you'll have an architecture where a user's recommendation set updates within seconds of their behavior changing. Architecture Overview The system has three layers: LayerServicesResponsibilityIngestKinesis Data Streams, Kinesis FirehoseCapture and fan-out user eventsProcess & ReasonLambda, Amazon Bedrock AgentEnrich events, build context, invoke LLMStore & ServeDynamoDB, S3Persist profiles, cache recs, store artifacts The key design decision is keeping the hot path (Kinesis → Lambda → Bedrock → DynamoDB) fully async and the serving path (API → DynamoDB cache) completely decoupled. The user never waits for Bedrock to respond; they get the last cached recommendation set while a fresh one is already being computed in the background. Event Flow Here's what happens end to end when a user clicks on a product: The app publishes a user.interaction event to Kinesis Data StreamsKinesis fans the event out to two consumers: Lambda Processor and Kinesis FirehoseFirehose archives the raw event to S3 (cheap, durable, great for retraining later)Lambda enriches the event with user history from DynamoDB User Profiles, then invokes the Bedrock AgentThe Bedrock Agent reasons over the enriched context (recent events + profile + item catalog embeddings from S3) and writes a fresh recommendation set to DynamoDB Rec CacheThe client app reads recommendations from the cache via a lightweight Lambda API — no Bedrock latency in the hot path Code: Publishing Events to Kinesis This is your app-side producer. Keep it thin — just serialize and publish. Do all enrichment downstream. Python import boto3 import json import uuid from datetime import datetime, timezone kinesis = boto3.client('kinesis', region_name='us-east-1') def publish_interaction(user_id: str, item_id: str, event_type: str, metadata: dict = {}): """ Publish a user interaction event to Kinesis Data Streams. Partition key is user_id so all events for a user land on the same shard. """ event = { 'event_id': str(uuid.uuid4()), 'user_id': user_id, 'item_id': item_id, 'event_type': event_type, # 'click', 'purchase', 'dwell', 'skip' 'timestamp': datetime.now(timezone.utc).isoformat(), 'metadata': metadata, } response = kinesis.put_record( StreamName='user-interactions', Data=json.dumps(event).encode('utf-8'), PartitionKey=user_id, # consistent routing per user ) return response['SequenceNumber'] # Example call publish_interaction( user_id='u_8821', item_id='prod_thriller_042', event_type='purchase', metadata={'price': 14.99, 'category': 'thriller', 'session_id': 'sess_xyz'} ) Tip: Use user_id as the partition key so all events for a given user land on the same shard and arrive in order. This matters when Lambda is building a recency-ordered event window. Code: Lambda Processor — Enrich and Invoke Bedrock This is the core of the pipeline. The Lambda reads from the Kinesis stream, pulls user context from DynamoDB, and invokes the Bedrock Agent with a structured prompt. Python import boto3 import json import os from datetime import datetime, timezone dynamodb = boto3.resource('dynamodb') bedrock = boto3.client('bedrock-agent-runtime', region_name='us-east-1') profiles_table = dynamodb.Table(os.environ['PROFILES_TABLE']) # DynamoDB User Profiles rec_table = dynamodb.Table(os.environ['REC_CACHE_TABLE']) # DynamoDB Rec Cache AGENT_ID = os.environ['BEDROCK_AGENT_ID'] AGENT_ALIAS = os.environ['BEDROCK_AGENT_ALIAS'] MAX_HISTORY = 20 # last N events to include in context def handler(event, context): for record in event['Records']: # Kinesis payload is base64-encoded payload = json.loads(record['kinesis']['data']) process_event(payload) def process_event(payload: dict): user_id = payload['user_id'] item_id = payload['item_id'] evt_type = payload['event_type'] # 1. Fetch user profile + recent history from DynamoDB response = profiles_table.get_item(Key={'user_id': user_id}) profile = response.get('Item', {'user_id': user_id, 'history': [], 'preferences': {}) # 2. Append current event and trim to window profile['history'].append({ 'item_id': item_id, 'event_type': evt_type, 'timestamp': payload['timestamp'], 'metadata': payload.get('metadata', {}), }) profile['history'] = profile['history'][-MAX_HISTORY:] # 3. Write enriched profile back profiles_table.put_item(Item=profile) # 4. Build prompt for Bedrock Agent prompt = build_personalization_prompt(profile) # 5. Invoke Bedrock Agent agent_response = bedrock.invoke_agent( agentId=AGENT_ID, agentAliasId=AGENT_ALIAS, sessionId=user_id, # session per user keeps conversational context inputText=prompt, ) # 6. Parse streaming response chunks recommendations = parse_agent_response(agent_response) # 7. Write to recommendation cache rec_table.put_item(Item={ 'user_id': user_id, 'recommendations': recommendations, 'generated_at': datetime.now(timezone.utc).isoformat(), 'ttl': int(datetime.now(timezone.utc).timestamp()) + 3600, # 1hr TTL }) def build_personalization_prompt(profile: dict) -> str: history_summary = '\n'.join([ f"- [{e['event_type'].upper()}] item={e['item_id']} category={e['metadata'].get('category','unknown')}" for e in profile['history'][-10:] ]) return f"""You are a real-time personalization agent. User profile: {json.dumps(profile.get('preferences', {}))} Recent interactions (most recent last): {history_summary} Based on this behavior, return exactly 5 personalized item recommendations as a JSON array. Each item must include: item_id, category, reasoning (1 sentence), confidence_score (0-1). Return only valid JSON. No explanation outside the JSON block.""" def parse_agent_response(agent_response) -> list: full_text = '' for event in agent_response['completion']: if 'chunk' in event: full_text += event['chunk']['bytes'].decode('utf-8') try: # Extract JSON from response start = full_text.index('[') end = full_text.rindex(']') + 1 return json.loads(full_text[start:end]) except (ValueError, json.JSONDecodeError): return [] Code: Serving Recommendations via Lambda API The serving layer never touches Bedrock. It reads purely from the DynamoDB cache, keeping p99 latency well under 10ms. Python import boto3 import json import os from datetime import datetime, timezone dynamodb = boto3.resource('dynamodb') rec_table = dynamodb.Table(os.environ['REC_CACHE_TABLE']) FALLBACK_RECS = ['popular_001', 'popular_002', 'popular_003'] # cold-start fallback def handler(event, context): user_id = event['pathParameters']['userId'] response = rec_table.get_item(Key={'user_id': user_id}) item = response.get('Item') if not item: # Cold start: user has no history yet return api_response(200, { 'user_id': user_id, 'recommendations': FALLBACK_RECS, 'source': 'fallback', 'generated_at': None, }) age_seconds = ( datetime.now(timezone.utc) - datetime.fromisoformat(item['generated_at']) ).total_seconds() return api_response(200, { 'user_id': user_id, 'recommendations': item['recommendations'], 'source': 'cache', 'generated_at': item['generated_at'], 'cache_age_sec': int(age_seconds), }) def api_response(status: int, body: dict) -> dict: return { 'statusCode': status, 'headers': { 'Content-Type': 'application/json', 'Access-Control-Allow-Origin': '*', }, 'body': json.dumps(body), } Service Comparison: Why Each AWS Service? ServiceWhy it's hereAlternative consideredKinesis Data StreamsOrdered, replayable, millisecond-latency fan-outSQS (no ordering guarantee per user), EventBridge (higher latency)Kinesis FirehoseZero-code delivery to S3 for archivingWriting to S3 directly in Lambda (adds failure surface)LambdaEvent-driven, scales to 0, tight Kinesis integrationECS Fargate (overkill for stateless enrichment)Amazon BedrockManaged LLM with agent runtime, no infra to maintainSelf-hosted model on SageMaker (more control, much more ops)DynamoDBSingle-digit ms reads, TTL support, scales automaticallyRDS (too slow for hot path), ElastiCache (extra cost for separate store)S3Cheap durable archive + model artifact storeDynamoDB for raw events (expensive and unnecessary) Things to Watch in Production Bedrock latency is variable. Claude Sonnet typically responds in 1-4 seconds but can spike. Since recs are written async to cache, this doesn't affect user-facing latency, but it does affect freshness. Monitor bedrock:InvokeAgent duration in CloudWatch. Kinesis shard scaling. One shard handles 1MB/s write or 1000 records/s. At 10k active users, you'll need to plan shard count carefully. Use Enhanced Fan-Out if you have multiple Lambda consumers reading the same stream. DynamoDB TTL for cache eviction. The serving Lambda sets a 1-hour TTL on each rec entry. If Bedrock hasn't updated the cache in over an hour (e.g., Lambda errors), users fall back to the popular items list. Adjust TTL based on how stale you can tolerate. Cold start users. New users have no history, so the Bedrock prompt has nothing useful to reason over. I recommend a popularity-based fallback as shown in the serving Lambda, and switching to personalized recs after the user's first 3-5 interactions. Wrapping Up The pattern here is worth generalizing: keep the reasoning layer (Bedrock) fully off the hot serving path. Write results to a fast cache (DynamoDB), serve from the cache, and let the agent pipeline update it continuously in the background. This gives you the intelligence of an LLM-powered agent without the latency of one. The same pattern applies to fraud scoring, content moderation queues, ops alerting — anywhere you need a reasoning system that reacts to real-time streams without blocking the user experience. References Amazon Kinesis Data Streams Developer GuideAmazon Kinesis Data Firehose Developer GuideAmazon Bedrock Agent Runtime — Invoke Agent APIAWS Lambda — Using AWS Lambda with Amazon KinesisAmazon DynamoDB — Time to Live (TTL)Amazon S3 — Best practices for event-driven architecturesBuilding Agents with Amazon BedrockEvent-Driven Architecture on AWS — Whitepaper

By Jubin Abhishek Soni CORE