DZone Spotlight

In today’s evolving technological landscape, the shift from monolithic architectures to microservices is a strategic move for many businesses. This is particularly relevant in the domain of reimbursement calculation systems. As I mentioned in my previous article Part 1, let's explore how such a transition can be effectively managed. The Monolithic Challenge Imagine a scenario where you have a large-scale, monolithic system - possibly a bulky C# console application or an extensive SQL Server stored procedure. This system is tasked with performing reimbursement calculations, typically running overnight through a batch process scheduled in SQL Server. While functional, this monolithic approach often leads to challenges in scalability, flexibility, and maintenance. Moving to Microservices The objective of migrating to microservices is to decompose this large, complex system into smaller, more manageable components. The transition to a microservices architecture aims to leverage the cloud's advantages, including scalability, resource optimization, and cost-effectiveness. Steps for Migration 1. Understanding the System Begin by defining the data models from the existing monolithic application to understand its workflow, dependencies, and key components of the reimbursement calculation process. Source data for this system is thru 837 File which is a standardized electronic format for healthcare claim information. This file is extracted and data is generally loaded into a database through another loading process for the purpose of reimbursement calculation. For example, a few data models from the 837 file might look as below: C# public class Patient { public string Name { get; set; } public DateTime DateOfBirth { get; set; } public string Address { get; set; } public string Gender { get; set; } public string PatientId { get; set; } } public class Provider { public string Name { get; set; } public string Address { get; set; } public string NPI { get; set; } public string TaxID { get; set; } public string RenderingProvider { get; set; } } public class Claim { public string ControlNumber { get; set; } public DateTime ServiceFromDate { get; set; } public DateTime ServiceToDate { get; set; } public string TypeOfBill { get; set; } public string AdmissionType { get; set; } public string DischargeStatus { get; set; } public List<string> DiagnosisCodes { get; set; } public List<string> ProcedureCodes { get; set; } } public class Insurance { public string PayerName { get; set; } public string PayerAddress { get; set; } public string PayerId { get; set; } public string SubscriberInformation { get; set; } public string SubscriberId { get; set; } public string CoordinationOfBenefitsData { get; set; } } public class ServiceLine { public string RevenueCode { get; set; } public DateTime ServiceDate { get; set; } public int ServiceUnits { get; set; } public decimal ServiceCharges { get; set; } public List<string> ServiceModifiers { get; set; } } 2. Identifying Microservices Break down the monolithic process into smaller, logically separated services. Each microservice should represent a specific aspect of the reimbursement calculation, such as input validation, calculation logic, and output generation. In many cases, a healthcare reimbursement system can involve multiple microservices working together to provide end-to-end functionality. Here are some microservices that might be part of a comprehensive healthcare reimbursement system: For demonstration purposes, I will provide a simplified implementation of the Reimbursement Calculation Service. It is assumed that patient information, procedure details, and fee schedule data are retrieved from their respective microservices and passed as inputs to this service, Reimbursement.web layer: C# using Microsoft.AspNetCore.Mvc; using Reimbursement.Service; namespace Reimbursement.Controllers { [Route("api/[controller]")] [ApiController] public class ReimbursementController : ControllerBase { private IReimbursementService _reimbursementService; public ReimbursementController(IReimbursementService reimbursementService) { _reimbursementService = reimbursementService; } [HttpPost("calculate")] public ActionResult<decimal> CalculateExpectedReimbursement(Patient patient, Procedure procedure, FeeSchedule feeSchedule) { try { decimal expectedReimbursement = _reimbursementService.CalculateExpectedReimbursement(patient, procedure, feeSchedule); return Ok(expectedReimbursement); } catch (Exception ex) { return StatusCode(500, $"Internal server error: {ex.Message}"); } } } } Reimbursement service layer: C# using System; namespace Reimbursement.Service { public class ReimbursementService : IReimbursementService { public decimal CalculateExpectedReimbursement(Patient patient, Procedure procedure, FeeSchedule feeSchedule) { // Check if the patient and procedure exist if (patient == null || procedure == null) { throw new ArgumentNullException("Patient and Procedure must be provided."); } // Check if the feeSchedule exists if (feeSchedule == null) { throw new ArgumentNullException("FeeSchedule must be provided."); } // Calculate the expected reimbursement decimal expectedReimbursement = feeSchedule.Fee; // Basic reimbursement logic // You can add more complex reimbursement calculations here based on patient data and rules return expectedReimbursement; } } } The exact composition and architecture of microservices in a healthcare reimbursement system may vary based on the specific needs and scale of the application. The services listed above are examples of components that can be part of such a system, and they may interact with each other through APIs or message queues to perform end-to-end reimbursement processes. 3. Batch Processing in the Cloud Adapt the overnight batch processing to the cloud environment. This could involve leveraging cloud-native services for scheduled tasks, ensuring that the process is reliable and scalable. CalculationService can also be triggered through the user interface manually in case users need to rerun for specific accounts only so that this service can be reused in places other than batch processing. Conclusion Migrating a complex, monolithic reimbursement calculation system to microservices and deploying it in the cloud is a transformative step. This approach not only modernizes the system but also brings significant benefits in terms of scalability, resource utilization, and cost savings, aligning the system with modern cloud capabilities and business objectives.

The U.S. Securities and Exchange Commission (SEC) recently announced its new rules for public companies regarding cybersecurity risk management, strategy, governance, and incident exposure. Some requirements apply to this year—for example, disclosures for fiscal years ending December 15, 2023, or later have new annual reporting requirements. As a result, organizations are wondering about how these new rules impact them. In this post, we’ll help unpack the new rules, what they mean to you, and what your DevOps and DevSecOps teams might need to implement in response. Understanding the SEC Announcement In the press release, SEC Chair Gary Gensler has a quote that helps to summarize why the new rules are being implemented: “Currently, many public companies provide cybersecurity disclosure to investors. I think companies and investors alike, however, would benefit if this disclosure were made in a more consistent, comparable, and decision-useful way. Through helping to ensure that companies disclose material cybersecurity information, today’s rules will benefit investors, companies, and the markets connecting them.” Section I of the SEC’s Final Rule document helps us further understand the impetus for these new rules. Let’s summarize some of the key points: Cybersecurity incidents have been reported inconsistently. Cybersecurity incidents are also likely underreported. Economic activity is increasingly dependent on electronic systems that are susceptible to cyber risk. The frequency and financial impacts of cyberattacks are on the rise. With that in mind, it makes sense why the SEC would want to standardize how incidents are reported. Let’s take a closer look at some of the new rules’ specifics. What Is a “Material Cybersecurity Incident”? The new rules establish requirements for reporting material cybersecurity incidents. For tech folks, a phrase like “material cybersecurity incident” can be tough legalese to decipher. It’s clear that a breach compromising millions of sensitive records or costing tens of millions of dollars is material, but at what point is a breach not considered material? Fortunately, we have some precedents to help guide our interpretation. The concept of “materiality” has long been important regarding SEC regulations. Typically, the key characteristics of “materiality” are summarized by these two quotes from Justice Thurgood Marshall in a 1976 opinion indicating a fact is material if: There is a “substantial likelihood that a reasonable shareholder would consider it important in deciding how to vote.” A reasonable investor would see the fact as “having significantly altered the ‘total mix’ of information made available.” Note that no specific dollar amount makes an incident material. Additionally, multiple connected immaterial incidents can become material in context. One example that the SEC’s Final Rule cites includes a single threat actor engaging in multiple smaller but continuous attacks against an organization. For “cybersecurity incident,” the SEC defines the term for us: “an unauthorized occurrence on or conducted through a registrant’s information systems that jeopardizes the confidentiality, integrity, or availability of a registrant’s information systems or any information residing therein.” What Is the Timeframe for Reporting an Incident? Once an incident is deemed “material,” affected organizations have to file SEC Form 8-K within four business days. Some exceptions to this timeframe exist, and they’re related to national security reasons and a provision for omitting information related to national defense and foreign policy. Which Organizations Are Affected? Public companies are affected by the new SEC rules. That includes foreign private issuers (FPIs), a type of company registered outside of the U.S. but doing significant business inside the U.S. How Does an Organization Disclose This Information? The reporting requirements in the new SEC rules call out different forms that organizations must complete to disclose relevant information. They even include a requirement for disclosures to use the Inline eXtensible Business Reporting Language (Inline XBRL). The table below breaks down the key forms related to the new SEC rules and when required. Note that the forms aren’t new, but the new SEC rules have added new requirements related to cybersecurity disclosures. Form What is the form used for as it relates to the new rules? When is it required? 8-K To disclose information related to any material cybersecurity incident To describe the material aspects of the reported incident To describe the scope, nature, and timing of the incident To describe the likely impact, including any impact on finances and operations Item 1.05 of Form 8-K must be filed within four business days of a cybersecurity incident being deemed material 10-K To fulfill annual reporting requirements from Regulation S-K Item 106 To disclose processes for assessing, identifying, and managing cybersecurity risk To disclose effects or likely effects of cybersecurity threats and past incidents Describe the board of directors’ role in cybersecurity risk oversight Describe the role of management in the assessment and management of cybersecurity threats Annually 6-K Similar to Form 8-K, but for foreign private issuers After a material cybersecurity incident 20-F Similar to Form 10-K, but for foreign private issuers Annually How Is This Different From Other Standards in the Past? For publicly traded companies, rules and regulations are nothing new. Many organizations already face strict reporting requirements related to regulations such as HIPAA, PCI DSS, and SOX. Some of the most meaningful changes for publicly traded organizations include: Standardization in reporting requirements: Previously, cybersecurity incidents were reported with varying levels of detail and frequency. The new SEC rules standardize how (based on the forms) and when an organization must report incidents. Well-defined annual reporting updates: Publicly traded organizations now must report on cybersecurity practices and impact annually through Form 10-K or Form 20-F. Now, let’s get down to brass tacks. If you’re a CISO or a part of your organization’s Security, DevSecOps, or Governance Risk and Compliance team, what are the practical implications of these new rules? How Do the New SEC Rules Affect Your Cybersecurity Measures? The new SEC rules for public companies effectively create cybersecurity, disclosure, and governance requirements that organizations must address in their internal processes and policies. For example, the new rules mean that affected organizations must quickly detect and analyze cybersecurity incidents. Incident response and analysis capabilities need to be mature enough to enable disclosure of the “nature, scope, and timing” of the event for adequate disclosure. The emphasis on board and management involvement also creates governance requirements for organizations. This may increase C-suite support for cybersecurity initiatives that may have otherwise languished. As a result, this increased emphasis on governance may drive an increased focus on leveraging the right tactics and tools to enable effective detection, prevention, and disclosure of cybersecurity threats. What Tooling Can Help You Adhere to the New Rules? From a tooling perspective, the new SEC rules should drive organizations to focus on the following: Incident detection and prevention, include the ability to identify and mitigate vulnerabilities before they become full-blown incidents. Incident response, which covers the ability to recover from cybersecurity incidents and capture the relevant data to disclose its “nature, scope, and timing” to shareholders. Let’s look at some of the tools and practices most relevant to enabling incident detection, prevention, and response. Continuous Monitoring Continuous monitoring of IT infrastructure is essential to threat detection, root cause analysis, and incident response. With continuous monitoring platforms, enterprises can verify that adequate security controls are in place. That way, they can detect anomalies and improve MTTR if an incident occurs. SIEM A SIEM tool aggregates log data across an environment, enabling alerting, reporting, analysis, and data retention capabilities. When coupled with effective logging, SIEM platforms provide many of the capabilities organizations need to comply with the new SEC cybersecurity rules. For example, an SIEM platform can provide continuous log data monitoring, correlating alerts, and security events from across all of your enterprise security tools. It will help you investigate and respond to threats quickly. Some Cloud SIEM platforms can even extend traditional SIEM capabilities for modern enterprises with user and entity behavior analytics (UEBA) to detect unknown threats. Why Logging Is Essential for Cybersecurity Log management is a critical part of any organization’s cybersecurity toolkit. Logging helps enterprises capture and retain key security and compliance data, and it enables the alerting and analytics capabilities of tools such as intrusion prevention or intrusion detection systems (IPS/IDS) and SIEM platforms. Robust log centralization and management tools are essential for your cybersecurity posture. And, logging isn’t just relevant to the new SEC rules. Guidelines, requirements, and frameworks (such as FedRAMP, PCI DSS, ISO 27001, HIPAA, and GLBA) may include some logging and data retention requirements for an organization. Conclusion The new SEC rules help standardize cybersecurity incident disclosures and emphasize the importance of governance in addressing cybersecurity risk. For publicly traded organizations, these rules add specificity and structure to handling cybersecurity incidents and reporting on cybersecurity posture. The right tools, specifically platforms that enable effective logging and incident response, are essential to tying together a cybersecurity strategy that mitigates risk and enables adherence to the new rules.

By Alvin Lee CORE

The AIDocumentLibraryChat project uses the Spring AI project with OpenAI to search in a document library for answers to questions. To do that, Retrieval Augmented Generation is used on the documents. Retrieval Augmented Generation The process looks like this: The process looks like this: Upload Document Store Document in Postgresql DB. Split Document to create Embeddings. Create Embeddings with a call to the OpenAI Embedding Model. Store the Document Embeddings in the Postgresql Vector DB. Search Documents: Create Search Prompt Create Embedding of the Search Prompt with a call to the OpenAI Embedding Model. Query the Postgresql Vector DB for documents with nearest Embedding distances. Query Postgresql DB for Document. Create Prompt with the Search Prompt and the Document text chunk. Request an answer from GPT Model and show the answer based on the search prompt and the Document text chunk. Document Upload The uploaded document is stored in the database to have the source document of the answer. The document text has to be split in chunks to create embeddings per chunk. The embeddings are created by an embedding model of OpenAI and are a vectors with more than 1500 dimensions to represent the text chunk. The embedding is stored in an AI document with the chunk text and the id of the source document in the vector database. Document Search The document search takes the search prompt and uses the Open AI embedding model to turn it in an embedding. The embedding is used to search in the vector database for the nearest neighbor vector. That means that the embeddings of search prompt and the text chunk that have the biggest similarities. The id in the AIDocument is used to read the document of the relational database. With the Search Prompt and the text chunk of the AIDocument, the Document Prompt created. Then, the OpenAI GPT model is called with the prompt to create an answer based on Search Prompt and the document context. That causes the model to create answers that are closely based on the documents provided and improves the accuracy. The answer of the GPT model is returned and displayed with a link of the document to provide the source of the answer. Architecture The architecture of the project is built around Spring Boot with Spring AI. The Angular UI provides the user interface to show the document list, upload the documents and provide the Search Prompt with the answer and the source document. It communicates with the Spring Boot backend via the rest interface. The Spring Boot backend provides the rest controllers for the frontend and uses Spring AI to communicate with the OpenAI models and the Postgresql Vector database. The documents are stored with Jpa in the Postgresql Relational database. The Postgresql database is used because it combines the relational database and the vector database in a Docker image. Implementation Frontend The frontend is based on lazy loaded standalone components build with Angular. The lazy loaded standalone components are configured in the app.config.ts: TypeScript export const appConfig: ApplicationConfig = { providers: [provideRouter(routes), provideAnimations(), provideHttpClient()] }; The configuration sets the routes and enables the the http client and the animations. The lazy loaded routes are defined in app.routes.ts: TypeScript export const routes: Routes = [ { path: "doclist", loadChildren: () => import("./doc-list").then((mod) => mod.DOCLIST), }, { path: "docsearch", loadChildren: () => import("./doc-search").then((mod) => mod.DOCSEARCH), }, { path: "**", redirectTo: "doclist" }, ]; In 'loadChildren' the 'import("...").then((mod) => mod.XXX)' loads the the route lazily from the provided path and sets the exported routes defined in the 'mod.XXX' constant. The lazy loaded route 'docsearch' has the index.ts to export the constant: TypeScript export * from "./doc-search.routes"; That exports the doc-search.routes.ts: TypeScript export const DOCSEARCH: Routes = [ { path: "", component: DocSearchComponent, }, { path: "**", redirectTo: "" }, ]; It defines the routing to the 'DocSearchComponent'. The fileupload can be found in the DocImportComponent with the template doc-import.component.html: HTML <h1 mat-dialog-title i18n="@@docimportImportFile">Import file</h1> <div mat-dialog-content> <p i18n="@@docimportFileToImport">File to import</p> @if(uploading) { <div class="upload-spinner"><mat-spinner></mat-spinner></div> } @else { <input type="file" (change)="onFileInputChange($event)"> } @if(!!file) { <div> <ul> <li>Name: {{file.name}</li> <li>Type: {{file.type}</li> <li>Size: {{file.size} bytes</li> </ul> </div> } </div> <div mat-dialog-actions> <button mat-button (click)="cancel()" i18n="@@cancel">Cancel</button> <button mat-flat-button color="primary" [disabled]="!file || uploading" (click)="upload()" i18n="@@docimportUpload">Upload</button> </div> The fileupload is done with the '<input type="file" (change)="onFileInputChange($event)">' tag. It provides the upload feature and calls the 'onFileInputChange(...)' method after each upload. The 'Upload' button calls the 'upload()' method to send the file to the server on click. The doc-import.component.ts has methods for the template: TypeScript @Component({ selector: 'app-docimport', standalone: true, imports: [CommonModule,MatFormFieldModule, MatDialogModule,MatButtonModule, MatInputModule, FormsModule, MatProgressSpinnerModule], templateUrl: './doc-import.component.html', styleUrls: ['./doc-import.component.scss'] }) export class DocImportComponent { protected file: File | null = null; protected uploading = false; private destroyRef = inject(DestroyRef); constructor(private dialogRef: MatDialogRef<DocImportComponent>, @Inject(MAT_DIALOG_DATA) public data: DocImportComponent, private documentService: DocumentService) { } protected onFileInputChange($event: Event): void { const files = !$event.target ? null : ($event.target as HTMLInputElement).files; this.file = !!files && files.length > 0 ? files[0] : null; } protected upload(): void { if(!!this.file) { const formData = new FormData(); formData.append('file', this.file as Blob, this.file.name as string); this.documentService.postDocumentForm(formData) .pipe(tap(() => {this.uploading = true;}), takeUntilDestroyed(this.destroyRef)) .subscribe(result => {this.uploading = false; this.dialogRef.close();}); } } protected cancel(): void { this.dialogRef.close(); } } This is the standalone component with its module imports and the injected 'DestroyRef'. The 'onFileInputChange(...)' method takes the event parameter and stores its 'files' property in the 'files' constant. Then it checks for the first file and stores it in the 'file' component property. The 'upload()' method checks for the 'file' property and creates the 'FormData()' for the file upload. The 'formData' constant has the datatype ('file'), the content ('this.file') and the filename ('this.file.name') appended. Then the 'documentService' is used to post the 'FormData()' object to the server. The 'takeUntilDestroyed(this.destroyRef)' function unsubscribes the Rxjs pipeline after the component is destroyed. That makes unsubscribing pipelines very convenient in Angular. Backend The backend is a Spring Boot application with the Spring AI framework. Spring AI manages the requests to the OpenAI models and the Vector Database Requests. Liquibase Database setup The database setup is done with Liquibase and the script can be found in the db.changelog-1.xml: XML <databaseChangeLog xmlns="http://www.liquibase.org/xml/ns/dbchangelog" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-3.8.xsd"> <changeSet id="1" author="angular2guy"> <sql>CREATE EXTENSION if not exists hstore;</sql> </changeSet> <changeSet id="2" author="angular2guy"> <sql>CREATE EXTENSION if not exists vector;</sql> </changeSet> <changeSet id="3" author="angular2guy"> <sql>CREATE EXTENSION if not exists "uuid-ossp";</sql> </changeSet> <changeSet author="angular2guy" id="4"> <createTable tableName="document"> <column name="id" type="bigint"> <constraints primaryKey="true"/> </column> <column name="document_name" type="varchar(255)"> <constraints notNullConstraintName="document_document_name_notnull" nullable="false"/> </column> <column name="document_type" type="varchar(25)"> <constraints notNullConstraintName="document_document_type_notnull" nullable="false"/> </column> <column name="document_content" type="blob"/> </createTable> </changeSet> <changeSet author="angular2guy" id="5"> <createSequence sequenceName="document_seq" incrementBy="50" startValue="1000" /> </changeSet> <changeSet id="6" author="angular2guy"> <createTable tableName="vector_store"> <column name="id" type="uuid" defaultValueComputed="uuid_generate_v4 ()"> <constraints primaryKey="true"/> </column> <column name="content" type="text"/> <column name="metadata" type="json"/> <column name="embedding" type="vector(1536)"> <constraints notNullConstraintName= "vectorstore_embedding_type_notnull" nullable="false"/> </column> </createTable> </changeSet> <changeSet id="7" author="angular2guy"> <sql>CREATE INDEX vectorstore_embedding_index ON vector_store USING HNSW (embedding vector_cosine_ops);</sql> </changeSet> </databaseChangeLog> In the changeset 4 the table for the Jpa document entity is created with the primary key 'id'. The content type/size is unknown and because of that set to 'blob'. I changeset 5 the sequence for the Jpa entity is created with the default properties of Hibernate 6 sequences that are used by Spring Boot 3.x. In changeset 6 the table 'vector_store' is created with a primary key 'id' of type 'uuid' that is created by the 'uuid-ossp' extension. The column 'content' is of type 'text'('clob' in other databases) to has a flexible size. The 'metadata' column stores the metadata in a 'json' type for the AIDocuments. The 'embedding' column stores the embedding vector with the number of OpenAI dimensions. In changeset 7 the index for the fast search of the 'embeddings' column is set. Due to limited parameters of the Liquibase '<createIndex ...>' '<sql>' is used directly to create it. Spring Boot / Spring AI implementation The DocumentController for the frontend looks like this: Java @RestController @RequestMapping("rest/document") public class DocumentController { private final DocumentMapper documentMapper; private final DocumentService documentService; public DocumentController(DocumentMapper documentMapper, DocumentService documentService) { this.documentMapper = documentMapper; this.documentService = documentService; } @PostMapping("/upload") public long handleDocumentUpload( @RequestParam("file") MultipartFile document) { var docSize = this.documentService .storeDocument(this.documentMapper.toEntity(document)); return docSize; } @GetMapping("/list") public List<DocumentDto> getDocumentList() { return this.documentService.getDocumentList().stream() .flatMap(myDocument ->Stream.of(this.documentMapper.toDto(myDocument))) .flatMap(myDocument -> { myDocument.setDocumentContent(null); return Stream.of(myDocument); }).toList(); } @GetMapping("/doc/{id}") public ResponseEntity<DocumentDto> getDocument( @PathVariable("id") Long id) { return ResponseEntity.ofNullable(this.documentService .getDocumentById(id).stream().map(this.documentMapper::toDto) .findFirst().orElse(null)); } @GetMapping("/content/{id}") public ResponseEntity<byte[]> getDocumentContent( @PathVariable("id") Long id) { var resultOpt = this.documentService.getDocumentById(id).stream() .map(this.documentMapper::toDto).findFirst(); var result = resultOpt.stream().map(this::toResultEntity) .findFirst().orElse(ResponseEntity.notFound().build()); return result; } private ResponseEntity<byte[]> toResultEntity(DocumentDto documentDto) { var contentType = switch (documentDto.getDocumentType()) { case DocumentType.PDF -> MediaType.APPLICATION_PDF; case DocumentType.HTML -> MediaType.TEXT_HTML; case DocumentType.TEXT -> MediaType.TEXT_PLAIN; case DocumentType.XML -> MediaType.APPLICATION_XML; default -> MediaType.ALL; }; return ResponseEntity.ok().contentType(contentType) .body(documentDto.getDocumentContent()); } @PostMapping("/search") public DocumentSearchDto postDocumentSearch(@RequestBody SearchDto searchDto) { var result = this.documentMapper .toDto(this.documentService.queryDocuments(searchDto)); return result; } } The 'handleDocumentUpload(...)' handles the uploaded file with the 'documentService' at the '/rest/document/upload' path. The 'getDocumentList()' handles the get requests for the document lists and removes the document content to save on the response size. The 'getDocumentContent(...)' handles the get requests for the document content. It loads the document with the 'documentService' and maps the 'DocumentType' to the 'MediaType'. Then it returns the content and the content type, and the browser opens the content based on the content type. The 'postDocumentSearch(...)' method puts the request content in the 'SearchDto' object and returns the AI generated result of the 'documentService.queryDocuments(...)' call. The method 'storeDocument(...)' of the DocumentService looks like this: Java public Long storeDocument(Document document) { var myDocument = this.documentRepository.save(document); Resource resource = new ByteArrayResource(document.getDocumentContent()); var tikaDocuments = new TikaDocumentReader(resource).get(); record TikaDocumentAndContent(org.springframework.ai.document.Document document, String content) { } var aiDocuments = tikaDocuments.stream() .flatMap(myDocument1 -> this.splitStringToTokenLimit( myDocument1.getContent(), CHUNK_TOKEN_LIMIT) .stream().map(myStr -> new TikaDocumentAndContent(myDocument1, myStr))) .map(myTikaRecord -> new org.springframework.ai.document.Document( myTikaRecord.content(), myTikaRecord.document().getMetadata())) .peek(myDocument1 -> myDocument1.getMetadata() .put(ID, myDocument.getId().toString())).toList(); LOGGER.info("Name: {}, size: {}, chunks: {}", document.getDocumentName(), document.getDocumentContent().length, aiDocuments.size()); this.documentVsRepository.add(aiDocuments); return Optional.ofNullable(myDocument.getDocumentContent()).stream() .map(myContent -> Integer.valueOf(myContent.length).longValue()) .findFirst().orElse(0L); } private List<String> splitStringToTokenLimit(String documentStr, int tokenLimit) { List<String> splitStrings = new ArrayList<>(); var tokens = new StringTokenizer(documentStr).countTokens(); var chunks = Math.ceilDiv(tokens, tokenLimit); if (chunks == 0) { return splitStrings; } var chunkSize = Math.ceilDiv(documentStr.length(), chunks); var myDocumentStr = new String(documentStr); while (!myDocumentStr.isBlank()) { splitStrings.add(myDocumentStr.length() > chunkSize ? myDocumentStr.substring(0, chunkSize) : myDocumentStr); myDocumentStr = myDocumentStr.length() > chunkSize ? myDocumentStr.substring(chunkSize) : ""; } return splitStrings; } The 'storeDocument(...)' method saves the document to the relational database. Then, the document is converted in a 'ByteArrayResource' and read with the 'TikaDocumentReader' of Spring AI to turn it in a AIDocument list. Then the AIDocument list is flatmapped to split the documents into chunks with the the 'splitToTokenLimit(...)' method that are turned in new AIDocuments with the 'id' of the stored document in the Metadata map. The 'id' in the Metadata enables loading the matching document entity for the AIDocuments. Then the embeddings for the AIDocuments are created implicitly with calls to the 'documentVsRepository.add(...)' method that calls the OpenAI Embedding model and stores the AIDocuments with the embeddings in the vector database. Then the result is returned. The method 'queryDocument(...)' looks like this: Java public AiResult queryDocuments(SearchDto searchDto) { var similarDocuments = this.documentVsRepository .retrieve(searchDto.getSearchString()); var mostSimilar = similarDocuments.stream() .sorted((myDocA, myDocB) -> ((Float) myDocA.getMetadata().get(DISTANCE)) .compareTo(((Float) myDocB.getMetadata().get(DISTANCE)))).findFirst(); var documentChunks = mostSimilar.stream().flatMap(mySimilar -> similarDocuments.stream().filter(mySimilar1 -> mySimilar1.getMetadata().get(ID).equals( mySimilar.getMetadata().get(ID)))).toList(); Message systemMessage = switch (searchDto.getSearchType()) { case SearchDto.SearchType.DOCUMENT -> this.getSystemMessage( documentChunks, (documentChunks.size() <= 0 ? 2000 : Math.floorDiv(2000, documentChunks.size()))); case SearchDto.SearchType.PARAGRAPH -> this.getSystemMessage(mostSimilar.stream().toList(), 2000); }; UserMessage userMessage = new UserMessage(searchDto.getSearchString()); Prompt prompt = new Prompt(List.of(systemMessage, userMessage)); LocalDateTime start = LocalDateTime.now(); AiResponse response = aiClient.generate(prompt); LOGGER.info("AI response time: {}ms", ZonedDateTime.of(LocalDateTime.now(), ZoneId.systemDefault()).toInstant().toEpochMilli() - ZonedDateTime.of(start, ZoneId.systemDefault()).toInstant() .toEpochMilli()); var documents = mostSimilar.stream().map(myGen -> myGen.getMetadata().get(ID)).filter(myId -> Optional.ofNullable(myId).stream().allMatch(myId1 -> (myId1 instanceof String))).map(myId -> Long.parseLong(((String) myId))) .map(this.documentRepository::findById) .filter(Optional::isPresent) .map(Optional::get).toList(); return new AiResult(searchDto.getSearchString(), response.getGenerations(), documents); } private Message getSystemMessage( List<org.springframework.ai.document.Document> similarDocuments, int tokenLimit) { String documents = similarDocuments.stream() .map(entry -> entry.getContent()) .filter(myStr -> myStr != null && !myStr.isBlank()) .map(myStr -> this.cutStringToTokenLimit(myStr, tokenLimit)) .collect(Collectors.joining("\n")); SystemPromptTemplate systemPromptTemplate = new SystemPromptTemplate(this.systemPrompt); Message systemMessage = systemPromptTemplate .createMessage(Map.of("documents", documents)); return systemMessage; } private String cutStringToTokenLimit(String documentStr, int tokenLimit) { String cutString = new String(documentStr); while (tokenLimit < new StringTokenizer(cutString, " -.;,").countTokens()){ cutString = cutString.length() > 1000 ? cutString.substring(0, cutString.length() - 1000) : ""; } return cutString; } The method first loads the documents best matching the 'searchDto.getSearchString()' from the vector database. To do that the OpenAI Embedding model is called to turn the search string into an embedding and with that embedding the vector database is queried for the AIDocuments with the lowest distance(the distance between the vectors of the search embedding and the database embedding). Then the AIDocument with the lowest distance is stored in the 'mostSimilar' variable. Then all the AIDocuments of the document chunks are collected by matching the document entity id of their Metadata 'id's. The 'systemMessage' is created with the 'documentChunks' or the 'mostSimilar' content. The 'getSystemMessage(...)' method takes them and cuts the contentChunks to a size that the OpenAI GPT models can handle and returns the 'Message'. Then the 'systemMessage' and the 'userMessage' are turned into a 'prompt' that is send with 'aiClient.generate(prompt)' to the OpenAi GPT model. After that the AI answer is available and the document entity is loaded with the id of the metadata of the 'mostSimilar' AIDocument. The 'AiResult' is created with the search string, the GPT answer, the document entity and is returned. The vector database repository DocumentVsRepositoryBean with the Spring AI 'VectorStore' looks like this: Java @Repository public class DocumentVSRepositoryBean implements DocumentVsRepository { private final VectorStore vectorStore; public DocumentVSRepositoryBean(JdbcTemplate jdbcTemplate, EmbeddingClient embeddingClient) { this.vectorStore = new PgVectorStore(jdbcTemplate, embeddingClient); } public void add(List<Document> documents) { this.vectorStore.add(documents); } public List<Document> retrieve(String query, int k, double threshold) { return new VectorStoreRetriever(vectorStore, k, threshold).retrieve(query); } public List<Document> retrieve(String query) { return new VectorStoreRetriever(vectorStore).retrieve(query); } } The repository has the 'vectorStore' property that is used to access the vector database. It is created in the constructor with the injected parameters with the 'new PgVectorStore(...)' call. The PgVectorStore class is provided as the Postgresql Vector database extension. It has the 'embeddingClient' to use the OpenAI Embedding model and the 'jdbcTemplate' to access the database. The method 'add(...)' calls the OpenAI Embedding model and adds AIDocuments to the vector database. The methods 'retrieve(...)' query the vector database for embeddings with the lowest distances. Conclusion Angular made the creation of the front end easy. The standalone components with lazy loading have made the initial load small. The Angular Material components have helped a lot with the implementation and are easy to use. Spring Boot with Spring AI has made the use of Large Language Models easy. Spring AI provides the framework to hide the creation of embeddings and provides an easy-to-use interface to store the AIDocuments in a vector database(several are supported). The creation of the embedding for the search prompt to load the nearest AIDocuments is also done for you and the interface of the vector database is simple. The Spring AI prompt classes make the creation of the prompt for the OpenAI GPT models also easy. Calling the model is done with the injected 'aiClient,' and the results are returned. Spring AI is a very good Framework from the Spring Team. There have been no problems with the experimental version. With Spring AI, the Large Language Models are now easy to use on our own documents.

Over the past four years, developers have harnessed the power of Quarkus, experiencing its transformative capabilities in evolving Java microservices from local development to cloud deployments. As we stand on the brink of a new era, Quarkus 3 beckons with a promise of even more enhanced features, elevating developer experience, performance, scalability, and seamless cloud integration. In this enlightening journey, let’s delve into the heart of Quarkus 3's integration with virtual threads (Project Loom). You will learn how Quarkus enables you to simplify the creation of asynchronous concurrent applications, leveraging virtual threads for unparalleled scalability while ensuring efficient memory usage and peak performance. Journey of Java Threads You might have some experience with various types of Java threads if you have implemented Java applications for years. Let me remind you real quick how Java threads have been evolving over the last decades. Java threads have undergone significant advancements since their introduction in Java 1.0. The initial focus was on establishing fundamental concurrency mechanisms, including thread management, thread priorities, thread synchronization, and thread communication. As Java matured, it introduced atomic classes, concurrent collections, the ExecutorService framework, and the Lock and Condition interfaces, providing more sophisticated and efficient concurrency tools. Java 8 marked a turning point with the introduction of functional interfaces, lambda expressions, and the CompletableFuture API, enabling a more concise and expressive approach to asynchronous programming. Additionally, the Reactive Streams API standardized asynchronous stream processing and Project Loom introduced virtual threads, offering lightweight threads and improved concurrency support. Java 19 further enhanced concurrency features with structured concurrency constructs, such as Flow and WorkStealing, providing more structured and composable concurrency patterns. These advancements have significantly strengthened Java's concurrency capabilities, making it easier to develop scalable and performant concurrent applications. Java threads continue to evolve, with ongoing research and development focused on improving performance, scalability, and developer productivity in concurrent programming. Virtual threads, generally available (GA) in Java 21, are a revolutionary concurrency feature that addresses the limitations of traditional operating system (OS) threads. OS threads are heavyweight, limited in scalability, and complex to manage, posing challenges for developing scalable and performant concurrent applications. Virtual threads also offer several benefits, such as being a lightweight and efficient alternative, consuming less memory, reducing context-switching overhead, and supporting concurrent tasks. They simplify thread management, improve performance, and enhance scalability, paving the way for new concurrency paradigms and enabling more efficient serverless computing and microservices architectures. Virtual threads represent a significant advancement in Java concurrency, poised to shape the future of concurrent programming. Getting Started With Virtual Threads In general, you need to create a virtual thread using Thread.Builder directly in your Java project using JDK 21. For example, the following code snippet showcases how developers can create a new virtual thread and print a message to the console from the virtual thread. The Thread.ofVirtual() method creates a new virtual thread builder, and the name() method sets the name of the virtual thread to "virtual-thread". Then, the start() method starts the virtual thread and executes the provided Runnable lambda expression, which prints a message to the console. Lastly, the join() method waits for the virtual thread to finish executing before continuing. The System.out.println() statement in the main thread prints a message to the console after the virtual thread has finished executing. Java public class MyVirtualThread { public static void main(String[] args) throws InterruptedException { // Create a new virtual thread using Thread.Builder Thread thread = Thread .ofVirtual() .name("my-vt") .start(() -> { System.out.println("Hello from virtual thread!"); }); // Wait for the virtual thread to finish executing thread.join(); System.out.println("Main thread completed."); } } Alternatively, you can implement the ThreadFactory interface to start a new virtual thread in your Java project with JDK 21. The following code snippet showcases how developers can define a VirtualThreadFactory class that implements the ThreadFactory interface. The newThread() method of this class creates a new virtual thread using the Thread.ofVirtual() method. The name() method of the Builder object is used to set the name of the thread and the factory() method is used to set the ThreadFactory object. Java // Implement a ThreadFactory to start a new virtual thread public class VirtualThreadFactory implements ThreadFactory { private final String namePrefix; public VirtualThreadFactory(String namePrefix) { this.namePrefix = namePrefix; } @Override public Thread newThread(Runnable r) { return Thread.ofVirtual() .name(namePrefix + "-" + r.hashCode()) .factory(this) .build(); } } You might feel it will get more complex when you try to run your actual methods or classes on top of the virtual threads. Luckily, Quarkus enables you to skip the learning curve and execute the existing blocking services on the virtual threads quickly and efficiently. Let’s dive into it. Quarkus Way to the Virtual Thread You just need to keep reminding yourself of two things to run an application on virtual threads. Implement blocking services rather than reactive (or non-blocking) services based on JDK 21. Use @RunOnVirtualThread annotation on top of a method or a class that you want. Here is a code snippet of how Quarkus allows you to run the process() method on a virtual thread. Java @Path("/hello") public class GreetingResource { @GET @Produces(MediaType.TEXT_PLAIN) @RunOnVirtualThread public String hello() { Log.info(Thread.currentThread()); return "Quarkus 3: The Future of Java Microservices with Virtual Threads and Beyond"; } } You can start the Quarkus Dev mode (Live coding) to verify the above sample application. Then, invoke the REST endpoint using the curl command. Shell $ curl http://localhost:8080/hello The output should look like this. Shell Quarkus 3: The Future of Java Microservices with Virtual Threads and Beyond When you take a look at the terminal, you see that Quarkus dev mode is running. You can see that a virtual thread is created to run this application. Shell (quarkus-virtual-thread-0) VirtualThread[#123,quarkus-virtual-thread-0]/runnable@ForkJoinPool-1-worker-1 Try to invoke the endpoint a few more times, and the logs in the terminal should look like this. You learned how Quarkus integrates the virtual thread for Java developers to run blocking applications with a single @RunOnVirtualThread annotation. You should be aware that this annotation is not a silver bullet for all use cases. In the next article, I’ll introduce pitfalls, limitations, and performance test results against reactive applications.

By Daniel Oh CORE

This guide delves into the meticulous steps of deploying a Spring MVC application on a local Tomcat server. This hands-on tutorial is designed to equip you with the skills essential for seamless deployment within your development environment. Follow along to enhance your proficiency in deploying robust and reliable Spring MVC apps, ensuring a smooth transition from development to production. Introduction In the preliminary stages, it's crucial to recognize the pivotal role of deploying a Spring MVC application on a local Tomcat server. This initial step holds immense significance as it grants developers the opportunity to rigorously test their applications within an environment closely mirroring the production setup. The emphasis on local deployment sets the stage for a seamless transition, ensuring that the application, when deemed ready for release, aligns effortlessly with the intricacies of the production environment. This strategic approach enhances reliability and mitigates potential challenges in the later stages of the development life cycle. Prerequisites To get started, ensure you have the necessary tools and software installed: Spring MVC Project: A well-structured Spring MVC project. Tomcat Server: Download and install Apache Tomcat, the popular servlet container. Integrated Development Environment (IDE): Use your preferred IDE (Eclipse, IntelliJ, etc.) for efficient development. Configuring the Spring MVC App Initiating the deployment process entails meticulous configuration of your Spring MVC app development project. Navigate to your project within the Integrated Development Environment (IDE) and focus on pivotal files such as `web.xml` and `dispatcher-servlet.xml` These files house crucial configurations that dictate the behavior of your Spring MVC application. Pay meticulous attention to details like servlet mappings and context configurations within these files. This configuration step is foundational, as it establishes the groundwork for the application's interaction with the servlet container, paving the way for a well-orchestrated deployment on the local Tomcat server. 1. Create the Spring Configuration Class In a typical Spring MVC application, you create a Java configuration class to define the application's beans and configuration settings. Let's call this class 'AppConfig'. Java import org.springframework.context.annotation.ComponentScan; import org.springframework.context.annotation.Configuration; import org.springframework.web.servlet.config.annotation.EnableWebMvc; @Configuration @EnableWebMvc @ComponentScan(basePackages = "com.example.controller") // Replace with your actual controller package public class AppConfig { // Additional configurations or bean definitions can go here } Explanation '@Configuration': Marks the class as a configuration class. '@EnableWebMvc': Enables Spring MVC features. '@ComponentScan': Scans for Spring components (like controllers) in the specified package. 2. Create the DispatcherServlet Configuration Create a class that extends 'AbstractAnnotationConfigDispatcherServletInitializer' to configure the DispatcherServlet. Java import org.springframework.web.servlet.support.AbstractAnnotationConfigDispatcherServletInitializer; public class MyWebAppInitializer extends AbstractAnnotationConfigDispatcherServletInitializer { @Override protected Class<?>[] getRootConfigClasses() { return null; // No root configuration for this example } @Override protected Class<?>[] getServletConfigClasses() { return new Class[]{AppConfig.class}; // Specify your configuration class } @Override protected String[] getServletMappings() { return new String[]{"/"}; } } Explanation 'getServletConfigClasses()' : Specifies the configuration class (in this case, AppConfig) for the DispatcherServlet. 'getServletMappings()' : Maps the DispatcherServlet to the root URL ("/"). Now, you've configured the basic setup for a Spring MVC application. This includes setting up component scanning, enabling MVC features, and configuring the DispatcherServlet. Adjust the package names and additional configurations based on your application's structure and requirements. Setting up Tomcat Server Locally Transitioning to the next phase involves the establishment of a local Tomcat server. Start by downloading the latest version of Apache Tomcat from the official website and meticulously follow the installation instructions. Once the installation process is complete, the next pivotal step is configuring Tomcat within your Integrated Development Environment (IDE). If you're using Eclipse, for example, seamlessly navigate to the server tab, initiate the addition of a new server, and opt for Tomcat from the available options. This localized setup ensures a synchronized and conducive environment for the impending deployment of your Spring MVC application. Building the Spring MVC App As you progress, it's imperative to verify that your Spring MVC project is poised for a seamless build. Leverage automation tools such as Maven or Gradle to expedite this process efficiently. Integrate the requisite dependencies into your project configuration file, such as the `pom.xml` for Maven users. Execute the build command to orchestrate the compilation and assembly of your project. This step ensures that your Spring MVC application is equipped with all the necessary components and dependencies, laying a solid foundation for subsequent phases of deployment on the local Tomcat server. 1. Project Structure Ensure that your project follows a standard Maven directory structure: CSS project-root │ ├── src │ ├── main │ │ ├── java │ │ │ └── com │ │ │ └── example │ │ │ ├── controller │ │ │ │ └── MyController.java │ │ │ └── AppConfig.java │ │ └── resources │ └── webapp │ └── WEB-INF │ └── views │ ├── pom.xml └── web.xml /* Write CSS Here */ 2. MyController.java: Sample Controller Create a simple controller that handles requests. This is a basic example; you can expand it based on your application requirements. Java package com.example.controller; import org.springframework.stereotype.Controller; import org.springframework.ui.Model; import org.springframework.web.bind.annotation.RequestMapping; @Controller public class MyController { @RequestMapping("/hello") public String hello(Model model) { model.addAttribute("message", "Hello, Spring MVC!"); return "hello"; // This corresponds to the view name } } 3. View ('hello.jsp') Create a simple JSP file under 'src/main/webapp/WEB-INF/views/hello.jsp' Java Server Pages <%@ page contentType="text/html;charset=UTF-8" language="java" %> <html> <head> <title>Hello Page</title> </head> <body> <h2>${message}</h2> </body> </html> 4. 'AppConfig.java': Configuration Ensure that AppConfig.java scans the package where your controllers are located Java package com.example; import org.springframework.context.annotation.ComponentScan; import org.springframework.context.annotation.Configuration; import org.springframework.web.servlet.config.annotation.EnableWebMvc; @Configuration @EnableWebMvc @ComponentScan(basePackages = "com.example.controller") public class AppConfig { // Additional configurations or bean definitions can go here } 5. 'web.xml': Web Application Configuration Configure the DispatcherServlet in web.xml XML <web-app xmlns="http://xmlns.jcp.org/xml/ns/javaee" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee http://xmlns.jcp.org/xml/ns/javaee/web-app_4_0.xsd" version="4.0"> <servlet> <servlet-name>dispatcher</servlet-name> <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class> <init-param> <param-name>contextConfigLocation</param-name> <param-value>/WEB-INF/dispatcher-servlet.xml</param-value> </init-param> <load-on-startup>1</load-on-startup> </servlet> <servlet-mapping> <servlet-name>dispatcher</servlet-name> <url-pattern>/</url-pattern> </servlet-mapping> </web-app> 6. 'dispatcher-servlet'.xml Create a dispatcher-servlet.xml file under src/main/webapp/WEB-INF/ to define additional Spring MVC configurations: XML <?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context" xmlns:mvc="http://www.springframework.org/schema/mvc" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc.xsd"> <!-- Enables component scanning for the specified package --> <context:component-scan base-package="com.example.controller"/> <!-- Enables annotation-driven Spring MVC --> <mvc:annotation-driven/> <!-- Resolves views selected for rendering by @Controllers to .jsp resources in the /WEB-INF/views directory --> <bean class="org.springframework.web.servlet.view.InternalResourceViewResolver"> <property name="prefix" value="/WEB-INF/views/"/> <property name="suffix" value=".jsp"/> </bean> </beans> 7. Run the Application Run your application (this depends on your IDE). Access the hello endpoint at http://localhost:8080/your-app-context/hello. You should see the "Hello, Spring MVC!" message. Remember to replace "your-app-context" with the actual context path of your deployed application. War File Creation Transitioning to the packaging phase, it's time to create a deployable Web Application Archive (WAR) file for your Spring MVC application. This file serves as the standardized encapsulation of your Java web application. Utilize prevalent build tools like Maven to automate this process, simplifying the generation of the WAR file. Typically, you'll find this compiled archive neatly organized within the target directory. The WAR file encapsulates your Spring MVC app, ready to be seamlessly deployed onto the local Tomcat server, marking a pivotal step towards actualizing the functionality of your application in a real-world web environment. Deploying on Tomcat Embarking on the deployment phase, the excitement builds as you launch your application onto the local Tomcat server. This involves a straightforward process: copy the previously generated WAR file into the designated `webapps` directory within your Tomcat installation. This directory serves as the portal for deploying web applications. Subsequently, initiate or restart the Tomcat server and watch as it autonomously detects and deploys your Spring MVC application. This automated deployment mechanism streamlines the process, ensuring that your application is swiftly up and running on the local Tomcat server, ready for comprehensive testing and further development iterations. Testing the Deployed App Upon successful deployment, it's time to conduct a comprehensive test of your Spring MVC application. Open your web browser and enter the address `http://localhost:8080/your-app-context`, replacing `your-app-context` with the precise context path assigned to your deployed application. This step allows you to visually inspect and interact with your application in a real-time web environment. If all configurations align seamlessly, you should witness your Spring MVC app dynamically come to life, marking a pivotal moment in the deployment process and affirming the correct integration of your application with the local Tomcat server. Tips for Efficient Development To enhance your development workflow, consider the following tips: Hot swapping: Leverage hot-swapping features in your IDE to avoid restarting the server after every code change. Logging: Implement comprehensive logging to troubleshoot any issues during deployment. Monitoring: Utilize tools like JConsole or VisualVM to monitor your application's performance metrics. Conclusion In reaching this conclusion, congratulations are in order! The successful deployment of your Spring MVC app on a local Tomcat server marks a significant milestone. This guide has imparted a foundational understanding of the deployment process, a vital asset for a seamless transition to production environments. As you persist in honing your development skills, bear in mind that adept deployment practices are instrumental in delivering applications of utmost robustness and reliability. Your achievement in this deployment endeavor underscores your capability to orchestrate a streamlined and effective deployment pipeline for future projects. Well done!

Apache Airflow is an open-source platform that allows you to programmatically author, schedule, and monitor workflows. It uses Python as its programming language and offers a flexible architecture suited for both small-scale and large-scale data processing. The platform supports the concept of Directed Acyclic Graphs to define workflows, making it easy to visualize complex data pipelines. One of the key features of Apache Airflow is its ability to schedule and trigger batch jobs, making it a popular choice for processing large volumes of data. It provides excellent support for integrating with various data processing technologies and frameworks such as Apache Hadoop and Apache Spark. By using Apache Airflow for batch processing, you can easily define and schedule your data processing tasks, ensuring that they are executed in the desired order and within the specified time constraints. Batch processing is a common approach in big data processing that involves the processing of data in large volumes, typically at regular time intervals. This approach is well-suited for scenarios where data can be collected over a period and processed together as a batch. Within the fintech sector, batch processing caters to a wide range of applications, including but not limited to authorization and settlement processes, management of recurring payments, enabling reconciliation operations, performing fraud detection and analytic tasks, adhering to regulatory mandates, and overseeing changes to customer relationship management systems. Let's explore a simple use case of processing an input file and writing back to the output file using Apache Airflow. To get started with Apache Airflow, you can follow the official documentation for installation and setup. Overview diagram illustrating the basic flow of a batch processing scenario Setting the Stage Python from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime, timedelta # Default arguments for the DAG default_args = { 'owner': 'airflow', 'start_date': datetime(2021, 1, 1), 'retries': 1, 'retry_delay': timedelta(minutes=5), } The script begins by importing necessary modules and defining default arguments for the DAG. These default parameters include the DAG owner, start date, and retry settings. Reading Function: Extracting Data Python def read_function(**kwargs): ti = kwargs["ti"] # Read from a file (example: input.txt) with open("path/to/file/input_file.txt", "r") as file: # Read the remaining lines lines = file.readlines() # Push each line to XCom storage for i, line in enumerate(lines): ti.xcom_push(key=f"line_{i}", value=line.strip()) # Push the total number of lines to XCom storage ti.xcom_push(key="num_lines", value=len(lines)) The read_function simulates the extraction of data by reading lines from a file (`input.txt`). It then uses Airflow's XCom feature to push each line and the total number of lines into storage, making it accessible to subsequent tasks. Sample Input File Plain Text CardNumber,TransactionId,Amount,TxnType,Recurring,Date 1,123456789,100.00,Debit,Monthly,2023-12-31 2,987654321,50.00,Credit,Weekly,2023-10-15 3,456789012,75.50,Debit,Monthly,2023-11-30 4,555111222,120.75,Credit,Daily,2023-09-30 In the given input file, we can see the handling of a recurring transactions file. Processing Function: Transforming Data Python def process_function(**kwargs): ti = kwargs["ti"] # Pull all lines from XCom storage lines = [ti.xcom_pull(task_ids="read", key=f"line_{i}") for i in range(ti.xcom_pull(task_ids="read", key="num_lines"))] # Process and print all lines for i, line in enumerate(lines): logging.info(f"Make Payment Transaction {i + 1}: {line}") The process_function pulls all lines from XCom storage and simulates the transformation process by printing each line to the console. This task demonstrates the flexibility of Airflow in handling data flow between tasks. The process_function can have multiple implementations, allowing it to either invoke a web service call to execute the transaction or call another DAG to follow a different flow. Logs Plain Text [2023-11-28, 03:49:06 UTC] {taskinstance.py:1662} INFO - Exporting env vars: AIRFLOW_CTX_DAG_OWNER='airflow' AIRFLOW_CTX_DAG_ID='batch_processing_dag' AIRFLOW_CTX_TASK_ID='process' AIRFLOW_CTX_EXECUTION_DATE='2023-11-28T03:48:00+00:00' AIRFLOW_CTX_TRY_NUMBER='1' AIRFLOW_CTX_DAG_RUN_ID='scheduled__2023-11-28T03:48:00+00:00' [2023-11-28, 03:49:06 UTC] {batch_processing_dag.py:38} INFO - Make Payment Transaction 1: 1,123456789,100.00,Debit,Monthly,2023-12-31 [2023-11-28, 03:49:06 UTC] {batch_processing_dag.py:38} INFO - Make Payment Transaction 2: 2,987654321,50.00,Credit,Weekly,2023-10-15 [2023-11-28, 03:49:06 UTC] {batch_processing_dag.py:38} INFO - Make Payment Transaction 3: 3,456789012,75.50,Debit,Monthly,2023-11-30 [2023-11-28, 03:49:06 UTC] {batch_processing_dag.py:38} INFO - Make Payment Transaction 4: 4,555111222,120.75,Credit,Daily,2023-09-30 [2023-11-28, 03:49:06 UTC] {python.py:194} INFO - Done. Returned value was: None Writing Function: Loading Data Python def write_function(**kwargs): ti = kwargs["ti"] # Pull all lines from XCom storage lines = [ti.xcom_pull(task_ids="read", key=f"line_{i}") for i in range(ti.xcom_pull(task_ids="read", key="num_lines"))] # Write all lines to an output file (example: output.txt) with open("path/to/file/processed.txt", "a") as file: for i, line in enumerate(lines): processed_line = f"{line.strip()} PROCESSED" file.write(f"{processed_line}\n") The write_function pulls all lines from XCom storage and writes them to an output file (`processed.txt`). Sample Output File After Transaction Is Processed Plain Text 1,123456789,100.00,Debit,Monthly,2023-12-31 PROCESSED 2,987654321,50.00,Credit,Weekly,2023-10-15 PROCESSED 3,456789012,75.50,Debit,Monthly,2023-11-30 PROCESSED 4,555111222,120.75,Credit,Daily,2023-09-30 PROCESSED DAG Definition: Orchestrating the Workflow Python dag = DAG( 'batch_processing_dag', default_args=default_args, description='DAG with Read, Process, and Write functions', schedule_interval='*/1 * * * *', # Set the schedule interval according to your needs catchup=False, ) The DAG is instantiated with the name batch_processing_dag, the previously defined default arguments, a description, a schedule interval (running every 1 minute), and the catchup parameter set to False. Task Definitions: Executing the Functions Python # Task to read from a file and push to XCom storage read_task = PythonOperator( task_id='read', python_callable=read_function, provide_context=True, dag=dag, ) # Task to process the data from XCom storage (print to console) process_task = PythonOperator( task_id='process', python_callable=process_function, provide_context=True, dag=dag, ) # Task to write the data back to an output file write_task = PythonOperator( task_id='write', python_callable=write_function, provide_context=True, dag=dag, ) Three tasks (read_task, process_task, and write_task) are defined using the PythonOperator. Each task is associated with one of the Python functions (read_function, process_function, and write_function). The provide_context=True parameter allows the functions to access the task instance and context information. Defining Task Dependencies Python # Define task dependencies read_task >> process_task >> write_task The task dependencies are specified using the >> operator, indicating the order in which the tasks should be executed. Conclusion In conclusion, Apache Airflow proves to be a flexible open-source tool that is great at managing workflows, especially when it comes to batch processing. It is the best choice for organizations of all sizes because it has features like dynamic workflow definition, support for Directed Acyclic Graphs (DAGs), careful task dependency management, full monitoring and logging, efficient parallel execution, and strong error handling. Illustrated by a straightforward batch processing scenario, the example emphasizes Apache Airflow's user-friendly interface and its adaptability to a range of data processing needs, showcasing its ease of use and versatility.

I had to recently add UI tests for an application implemented with Swing library for the Posmulten project. The GUI does not do any rocket science. It does what the Posmulten project was created for, generating DDL statements that make RLS policy for the Postgres database, but with a user interface based on Swing components. Now, because the posmulten is an open-source project and the CI/CD process uses GitHub action, it would be worth having tests covering the UI application's functionality. Tests that could be run in a headless environment. Testing Framework As for testing purposes, I picked the AssertJ Swing library. It is effortless to mimic application users' actions. Not to mention that I could, with no effort, check application states and their components. Below is an example of a simple test case that checks if the correct panel will show up with the expected content after entering text and clicking the correct button. Java @Test public void shouldDisplayCreationScriptsForCorrectConfigurationWhenClickingSubmitButton() throws SharedSchemaContextBuilderException, InvalidConfigurationException { // GIVEN String yaml = "Some yaml"; ISharedSchemaContext context = mock(ISharedSchemaContext.class); Mockito.when(factory.build(eq(yaml), any(DefaultDecoratorContext.class))).thenReturn(context); List<SQLDefinition> definitions = asList(sqlDef("DEF 1", null), sqlDef("ALTER DEFINIT and Function", null)); Mockito.when(context.getSqlDefinitions()).thenReturn(definitions); window.textBox(CONFIGURATION_TEXTFIELD_NAME).enterText(yaml); // WHEN window.button("submitBtn").click(); // THEN window.textBox(CREATION_SCRIPTS_TEXTFIELD_NAME).requireText("DEF 1" + "\n" + "ALTER DEFINIT and Function"); // Error panel should not be visible findJTabbedPaneFixtureByName(ERROR_TAB_PANEL_NAME).requireNotVisible(); } You can find the complete test code here. Posmulten The library for which the GUI application was created is generally a simple DDL statement builder that makes RSL policy in the Postgres database. The generated RLS policies allow applications communicating with the Postgres database to work in Mutli-tenant architecture with the shared schema strategy. For more info, please check below links: Posmulten GUI module Shared Schema Strategy With Postgres Multi-tenancy Architecture With Shared Schema Strategy in Webapp Application Based on Spring-boot, Thymeleaf, and Posmulten-hibernate Maven Configuration It is worth excluding UI tests from unit tests. Although tests might not be fully e2e with mocked components, it is worth excluding from running together with unit tests because their execution might take a little longer than running standard unit tests. XML <profile> <id>swing-tests</id> <activation> <activeByDefault>false</activeByDefault> </activation> <build> <plugins> <plugin> <groupId>org.codehaus.gmavenplus</groupId> <artifactId>gmavenplus-plugin</artifactId> <version>1.5</version> <executions> <execution> <goals> <goal>compile</goal> <goal>testCompile</goal> </goals> </execution> </executions> </plugin> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-surefire-plugin</artifactId> <version>2.22.1</version> <configuration> <includes> <include>**/*SwingTest.java</include> </includes> </configuration> </plugin> </plugins> </build> </profile> Full Maven file. To run tests locally, on an environment with a Graphics card, you need to execute tests with a maven wrapper, like below. Shell ./mvnw -pl :openwebstart '-DxvfbRunningTests=true' -P !unit-tests,swing-tests test GitHub Action Now, moving to the GitHub action, running the UI test on the environment with a Graphics card seems easy. However, there might be situations when some UI windows with a WhatsApp or MS Teams notification appear on the Desktop on which UI tests are executed, and our tests will fail. Tests should be repeated in such cases, but that is not the problem. Many more problems can occur when we try to execute tests on a headless environment, which is probably the default environment for every CI/CD pipeline. And we still need to run those tests and ensure they will pass no matter if they are executed in such an environment. When we ask how to execute UI tests in a headless environment, the first suggestion on the internet is to use the Xvfb. However, the contributors of AssertJ Swing suggest a different approach. Our tests maximize windows and do other stuff the default window manager of xvfb doesn't support. TightVNC makes it easy to use another window manager. Just add gnome-wm & (or the window manager of your choice) to ~/.vnc/xstartup and you're ready to run. GitHub So, I followed suggestions from the contributors' team and used the Tightvncserver. I had some problems with adding the gnome-wm. Instead, I used the Openbox. Below, you can see the step that runs UI tests. The full GitHub action file can be found here. The script files used to configure CI can be found here. YAML testing_swing_app: needs: [compilation_and_unit_tests, database_tests, testing_configuration_jar] runs-on: ubuntu-latest name: "Testing Swing Application" steps: - name: Git checkout uses: actions/checkout@v2 # Install JDKs and maven toolchain - uses: actions/setup-java@v3 name: Set up JDK 11 id: setupJava11 with: distribution: 'zulu' # See 'Supported distributions' for available options java-version: '11' - name: Set up JDK 1.8 id: setupJava8 uses: actions/setup-java@v1 with: java-version: 1.8 - uses: cactuslab/maven-toolchains-xml-action@v1 with: toolchains: | [ {"jdkVersion": "8", "jdkHome": "${{steps.setupJava8.outputs.path}"}, {"jdkVersion": "11", "jdkHome": "${{steps.setupJava11.outputs.path}"} ] - name: Install tightvncserver run: sudo apt-get update && sudo apt install tightvncserver - name: Install openbox run: sudo apt install openbox - name: Copy xstartup run: mkdir $HOME/.vnc && cp ./swing/xstartup $HOME/.vnc/xstartup && chmod +x $HOME/.vnc/xstartup - name: Setting password for tightvncserver run: ./swing/setpassword.sh - name: Run Swing tests id: swingTest1 continue-on-error: true run: ./mvnw -DskipTests --quiet clean install && ./swing/execute-on-vnc.sh ./mvnw -pl :openwebstart '-DxvfbRunningTests=true' -P !unit-tests,swing-tests test #https://www.thisdot.co/blog/how-to-retry-failed-steps-in-github-action-workflows/ # https://stackoverflow.com/questions/54443705/change-default-screen-resolution-on-headless-ubuntu - name: Run Swing tests (Second time) id: swingTest2 if: steps.swingTest1.outcome == 'failure' run: ./mvnw -DskipTests --quiet clean install && ./swing/execute-on-vnc.sh ./mvnw -pl :openwebstart '-DxvfbRunningTests=true' -P !unit-tests,swing-tests test Retry Failed Steps in GitHub Action Workflows As you probably saw in the GitHub action file, the last step, which executes the UI tests, is added twice. After correctly setting up Tightvncserver and Openbox, I didn't observe that the second step had to be executed, except at the beginning during deployment when there were not a lot of UI components. I used Xvfb, and sometimes, the CI passed only after the second step. So even if there is no problem with executing the test the first time right now, then it is still worth executing those tests the second time in case of failure. To check if the first step failed, we first have to name it. In this case, the name is "swingTest1". In the second step, we use the "if" property like the below: YAML if: steps.swingTest1.outcome == 'failure' And that is generally all for running the step a second time in case of failure. Check this resource if you want to check other ways to execute the step a second time. Summary Setting CI for UI tests might not be a trivial task, but it can benefit a project with any GUI. Not all things can be tested with unit tests.

By Szymon Tarnowski CORE

Have you ever wondered how data warehouses are different from Databases? And what are Data Lakes and Data Lake Houses? Let’s understand these with a hypothetical example. Bookster.biz is the new sensation in selling books worldwide. The business is flourishing, and they need to keep track of a lot of data: a large catalog of millions of books, millions of customers worldwide placing billions of orders to buy books. How do they keep track of all this data? How do they ensure their website and apps don’t grind to a halt because of all this load? Databases to the Rescue Databases are the workhorses of websites and mobile apps, handling all the data and millions of transactions. These databases come in many flavors (we will cover all different types of databases in a separate post). Still, the most popular ones are called Relational Databases (aka RDBMS), like MySQL, Postgres, Oracle, etc. Bookster would possibly have the following tables and schema (not exhaustive for brevity): BookCatalog: book ID, ISBN, title, authors, description, publisher, … BookInventory: book ID, number of books available for sale, ... Users: user ID, user name, email, … Orders: Order ID, book ID, user ID, payment information, order status, … When a user orders a book, Bookster will update two records simultaneously: reducing book inventory and inserting a new order entry in the Orders table. RDBMSs support transactions that enable such atomic operations where either all such operations succeed or all fail. Imagine if two or more users could order the last copy of a popular book. Without transaction support, all customers will place orders, and Bookster will have many pissed-off customers except one. Similarly, if the Database host crashes during the processing, the data may be inconsistent without transactions. This database interaction type is called Online Transaction Processing (aka OLTP), where the read and write operations happen very fast on a small amount of data, i.e., precisely two rows in the previous example. This is great. The customers are now happy, and they can order books fast. But the management wants to know what’s going on with the business. Which books are the best-sellers in different categories? Which authors are trending, and which are not selling much? How many orders are coming from which geographies or demographics? These kinds of answers are not accessible with just the databases. Data Warehouses Shine for Analytical Queries Data Warehouses (DWs) can handle large amounts of data, e.g., billions of orders, millions of book entries, etc. Bookster can load the data from the Database to the DW to answer the management questions. The analytical queries read a lot of data and summarise it in some form, like listing the total number of orders for a particular book broken down by geography and demographics. Examples of popular DWs are AWS Redshift, GCP BigQuery, etc. This database interaction type is called Online Analytical Processing (aka OLAP), where most reads happen on a large amount of data. The data is uploaded to the DWs in batches or can be streamed. The loading process is also known as ETL (Extract, Transform, and Load), which is done regularly to keep the DW in sync with the Database updates. DWs typically don't allow updating data but only add a newer version. Like RDBMS, DWs also have a notion of schema where tables and schema are well defined, and the ETL process converts the data into appropriate schema for loading. Some data doesn’t fit the schema easily but can be used by Machine Learning (ML) processes. For example, customers review different books as a text or a video review, and some rockstar ML engineers want to generate popular books by training an LLM on all books. So, the data can’t be structured as a strict schema anymore. Data Lakes help here by storing even more significant amounts of data with different formats and allowing efficient processing. Data Lakes and Data Lake Houses Are the Relatively New Kids on the Block Data Lakes (DLs) overcome the friction of converting the data into a specific format irrespective of if and when it will be used. Vast amounts of data in different native formats like JSON, text, binary, images, videos, etc., can be stored in a DL and converted to a specific schema at read time only when there is a need to process the data. The processing is flexible and scalable as DLs can support big data processing frameworks like Apache Spark. On the flip side, such flexibility could become a drawback if most of the data ingested is low quality due to the lack of data quality check or governance, making DL a ‘Data Swamp’ instead. That’s where the clever people of Databricks combined the goodness of DWs with DLs to create Data Lake Houses (DLHs). DLHs are more flexible than DWs, allowing schema both at the time of writing or reading, as needed, but with stricter mechanisms for data quality checks and metadata management, aka Data Governance. Also, DLHs allow flexibility in big data processing like DLs. The following table summarises the differences between these technologies: Key Characteristics Suitable for Drawbacks Examples Database Fast, small queries, transaction support Online use cases (OLTP) Not ideal for large analytical queries RDBMS: MySQL Data Warehouse Slow, large queries, no updates after write Analytics (OLAP) Less flexible as strict schema and lack of support for big data processing frameworks AWS Redshift, Google BigQuery, *Snowflake Data Lake Unstructured data, schema on read, flexible and big data processing Analytics (OLAP) Data quality issues due to lack of Data Governance *Snowflake, **AWS Lake Formation, **Databricks Delta Lake Data Lake House Structured or unstructured data, flexible with better Data Governance and supports big data processing Analytics (OLAP) More complex, less performance, and more expensive compared to DW *Snowflake, **AWS Lake Formation, **Databricks Delta Lake *Snowflake can be configured as a Data Warehouse, Data Lake, or Data Lake House. **AWS Lake Formation and Databricks Delta Lake can be configured as either Data Lake or Data Lake House.

Java, known for its versatility and robustness, has often faced criticism for its verbosity. However, it's essential to recognize that Java's perceived verbosity is not always a fault of the language itself but can be attributed to overengineering in code design. In this article, we'll explore the benefits of simplifying Java code by reducing unnecessary layers and interfaces and unlocking the power of simplicity for enhanced maintainability without sacrificing functionality. The Pitfall of Unnecessary Interfaces One common practice contributing to code complexity is the creation of interfaces without a clear purpose. Consider the classical case of having one interface for one implementation: Java public interface CreditCard { String payment(); } public class CreditCardImpl implements CreditCard{ String payment(); } The first sign of an unnecessary interface is the generation of a non-meaningful name, going against the principles of Clean Code advocated by Robert Martin. Instead of creating separate interfaces and implementations, a more straightforward approach is to have a single class handling both: Java public class CreditCard { public String payment() { return "Payment done!"; } } By eliminating the unnecessary interface, the code becomes more concise and adheres to the principles of clarity and simplicity. Choosing Interfaces Wisely Interfaces are potent tools in Java, but they should be used judiciously. One valid interface use case is implementing design patterns like the strategy pattern. For instance, you might have various strategies in a payment system, such as credit card payments, debit card payments, and more. In such scenarios, interfaces can help define a common contract: Java public interface Payment { String payment(); } public class CreditCard implements Payment { public String payment() { return "Credit card payment done!"; } } public class DebitCard implements Payment { public String payment() { return "Debit card payment done!"; } } Here, interfaces provide a unified structure for different payment strategies. The Unnecessary Layer Conundrum Another pitfall in code design involves the creation of unnecessary layers that act as mere pass-throughs, adding complexity without offering tangible benefits. Consider a scenario where an additional layer is introduced without any clear purpose: Java public class PaymentGateway { private CreditCard creditCard; public PaymentGateway(CreditCard creditCard) { this.creditCard = creditCard; } public String processPayment() { // Some processing logic return creditCard.payment(); } } In cases where the added layer serves no meaningful purpose, it's advisable to remove it, simplifying the code and improving its clarity: Java public class PaymentProcessor { private CreditCard creditCard; public PaymentProcessor(CreditCard creditCard) { this.creditCard = creditCard; } public String processPayment() { // Processing logic directly in the class return creditCard.payment(); } } Eliminating unnecessary layers makes the code more straightforward to maintain. Embracing Simplicity for Maintainability In conclusion, the key to unlocking the full potential of Java lies in embracing simplicity. Avoid unnecessary interfaces and layers that add complexity without providing clear benefits. Choose interfaces wisely, leveraging them for scenarios that enhance code structure, such as implementing design patterns. By simplifying your Java code, you make it more readable and maintainable, ensuring a more efficient and enjoyable development process. Video

By Otavio Santana CORE

In this article, I will show you how to use Cloudera DataFlow powered by Apache NiFi to interact with IBM WatsonX.AI foundation large language models in real time. We can work with any of the foundation models such as Google FLAN T5 XXL or IBM Granite models. I’ll show you how easy it is to build a real-time data pipeline feeding your Slack-like and mobile applications questions directly to secure WatsonX.AI models running in IBM Cloud. We will handle all the security, management, lineage, and governance with Cloudera Data Flow. As part of decision-making, we can choose different WatsonX.AI models on the fly based on what type of prompt it is. For example, if we want to continue a sentence versus answering a question I can pick different models. For questions answering Google FLAN T5 XXL works well. If I want to continue sentences I would use one of the IBM Granite models. You will notice how amazingly fast the WatsonX.AI models return the results we need. I do some quick enrichment and transformation and then send them out their way to Cloudera Apache Kafka to be used for continuous analytics and distribution to many other applications, systems, platforms, and downstream consumers. We will also output our answers to the original requester which could be someone in a Slack channel or someone in an application. All of this happens in real-time, with no code, full governance, lineage, data management, and security at any scale and on any platform. The power of IBM and Cloudera together in private, public, and hybrid cloud environments for real-time data and AI is just getting started. Try it today. Step By Step Real-Time Flow First, in Slack, I type a question: “Q: What is a good way to integrate Generative AI and Apache NiFi?” NiFi Flow Top Once that question is typed, the Slack server sends these events to our registered service. This can be hosted anywhere publicly facing. (Click here for Slack API link) Slack API Once enabled, your server will start receiving JSON events for each Slack post. This is easy to receive and parse in NiFi. Cloudera DataFlow enables receiving secure HTTPS REST calls in the public cloud-hosted edition with ease, even in Designer mode. NiFi Top Flow 2 In the first part of the flow, we received the REST JSON Post, which is as follows. Slackbot 1.0 (+https://api.slack.com/robots) application/json POST HTTP/1.1 { "token" : "qHvJe59yetAp1bao6wmQzH0C", "team_id" : "T1SD6MZMF", "context_team_id" : "T1SD6MZMF", "context_enterprise_id" : null, "api_app_id" : "A04U64MN9HS", "event" : { "type" : "message", "subtype" : "bot_message", "text" : "==== NiFi to IBM <http://WatsonX.AI|WatsonX.AI> LLM Answers\n\nOn Date: Wed, 15 Nov 20 This is a very rich detailed JSON file that we could push immediately raw to an Apache Iceberg Open Cloud Lakehouse, a Kafka topic, or an object store as a JSON document (Enhancement Option). I am just going to parse what I need. EvaluateJSONPath We parse out the channel ID and plain text of the post. I only want messages from general (“C1SD6N197”). Then I copy the texts to an inputs field as is required for Hugging Face. We check our input: if it’s stocks or weather (more to come) we avoid calling the LLM. SELECT * FROM FLOWFILE WHERE upper(inputs) like '%WEATHER%' AND not upper(inputs) like '%LLM SKIPPED%' SELECT * FROM FLOWFILE WHERE upper(inputs) like '%STOCK%' AND not upper(inputs) like '%LLM SKIPPED%' SELECT * FROM FLOWFILE WHERE (upper(inputs) like 'QUESTION:%' OR upper(inputs) like 'Q:%') and not upper(inputs) like '%WEATHER%' and not upper(inputs) like '%STOCK%' For Stocks processing: To parse what stock we need I am using my Open NLP processor to get it. So you will need to download the processor and the Entity extraction models. GitHub - tspannhw/nifi-nlp-processor: Apache NiFi NLP Processor Open NLP Example Apache NiFi Processor Then we pass that company name to an HTTP REST endpoint from AlphaVantage that converts the Company Name to Stock symbols. In free accounts, you only get a few calls a day, so if we fail we then bypass this step and try to just use whatever you passed in. Using RouteOnContent we filter an Error message out. Then we use a QueryRecord processor to convert from CSV to JSON and filter. SELECT name as companyName, symbol FROM FLOWFILE ORDER BY matchScore DESC LIMIT 1 We do a SplitRecord to ensure we are only one record. We then run EvaluateJsonPath to get our fields as attributes. In an UpdateAttribute we trim the symbol just in case. ${stockSymbol:trim()} We then pass that stock symbol to Twelve Data via InvokeHTTP to get our stock data. We then get a lot of stock data back. { "meta" : { "symbol" : "IBM", "interval" : "1min", "currency" : "USD", "exchange_timezone" : "America/New_York", "exchange" : "NYSE", "mic_code" : "XNYS", "type" : "Common Stock" }, "values" : [ { "datetime" : "2023-11-15 10:37:00", "open" : "152.07001", "high" : "152.08000", "low" : "151.99500", "close" : "152.00999", "volume" : "8525" }, { "datetime" : "2023-11-15 10:36:00", "open" : "152.08501", "high" : "152.12250", "low" : "152.08000", "close" : "152.08501", "volume" : "15204" } ... We then run EvaluateJSONPath to grab the exchange information. We fork the record to just get one record as this is just to return to Slack. We use UpdateRecord calls to enrich the stock data with other values. We then run a QueryRecord to limit us to 1 record to send to Slack. SELECT * FROM FLOWFILE ORDER BY 'datetime' DESC LIMIT 1 We run an EvaluateJsonPath to get the most value fields to display. We then run a PutSlack with our message. LLM Skipped. Stock Value for ${companyName} [${nlp_org_1}/${stockSymbol}] on ${date} is ${closeStockValue}. stock date ${stockdateTime}. stock exchange ${exchange} We also have a separate flow that is split from Company Name. In the first step, we call Yahoo Finance to get RSS headlines for that stock. https://feeds.finance.yahoo.com/rss/2.0/headline?s=${stockSymbol:trim()}&region=US&lang=en-US We use QueryRecord to convert RSS/XML Records to JSON. We then run a SplitJSON to break out the news items. We run a SplitRecord to limit to 1 record. We use EvaluateJSONPath to get the fields we need for our Slack message. We then run UpdateRecord to finalize our JSON. We then send this message to Slack. LLM Skipped. Stock News Information for ${companyName} [${nlp_org_1}/${stockSymbol}] on ${date} ${title} : ${description}. ${guid} article date ${pubdate} For those who selected weather, we follow a similar route (we should add caching with Redis @ Aiven) to stocks. We use my OpenNLP processor to extract locations you might want to have weather on. The next step is taking the output of the processor and building a value to send to our Geoencoder. weatherlocation = ${nlp_location_1:notNull():ifElse(${nlp_location_1}, "New York City")} If we can’t find a valid location, I am going to say “New York City." We could use some other lookup. I am doing some work on loading all locations and could do some advanced PostgreSQL searches on that - or perhaps OpenSearch or a vectorized datastore. I pass that location to Open Meteo to find the geo via InvokeHTTP. https://geocoding-api.open-meteo.com/v1/search?name=${weatherlocation:trim():urlEncode()}&count=1&language=en&format=json We then parse the values we need from the results. { "results" : [ { "id" : 5128581, "name" : "New York", "latitude" : 40.71427, "longitude" : -74.00597, "elevation" : 10.0, "feature_code" : "PPL", "country_code" : "US", "admin1_id" : 5128638, "timezone" : "America/New_York", "population" : 8175133, "postcodes" : [ "10001", "10002", "10003", "10004", "10005", "10006", "10007", "10008", "10009", "10010", "10011", "10012", "10013", "10014", "10016", "10017", "10018", "10019", "10020", "10021", "10022", "10023", "10024", "10025", "10026", "10027", "10028", "10029", "10030", "10031", "10032", "10033", "10034", "10035", "10036", "10037", "10038", "10039", "10040", "10041", "10043", "10044", "10045", "10055", "10060", "10065", "10069", "10080", "10081", "10087", "10090", "10101", "10102", "10103", "10104", "10105", "10106", "10107", "10108", "10109", "10110", "10111", "10112", "10113", "10114", "10115", "10116", "10117", "10118", "10119", "10120", "10121", "10122", "10123", "10124", "10125", "10126", "10128", "10129", "10130", "10131", "10132", "10133", "10138", "10150", "10151", "10152", "10153", "10154", "10155", "10156", "10157", "10158", "10159", "10160", "10161", "10162", "10163", "10164", "10165", "10166", "10167", "10168", "10169", "10170", "10171", "10172", "10173", "10174", "10175", "10176", "10177", "10178", "10179", "10185", "10199", "10203", "10211", "10212", "10213", "10242", "10249", "10256", "10258", "10259", "10260", "10261", "10265", "10268", "10269", "10270", "10271", "10272", "10273", "10274", "10275", "10276", "10277", "10278", "10279", "10280", "10281", "10282", "10285", "10286" ], "country_id" : 6252001, "country" : "United States", "admin1" : "New York" } ], "generationtime_ms" : 0.92196465 } We then parse the results so we can call another API to get the current weather for that latitude and longitude via InvokeHTTP. https://api.weather.gov/points/${latitude:trim()},${longitude:trim()} The results are geo-json. { "@context": [ "https://geojson.org/geojson-ld/geojson-context.jsonld", { "@version": "1.1", "wx": "https://api.weather.gov/ontology#", "s": "https://schema.org/", "geo": "http://www.opengis.net/ont/geosparql#", "unit": "http://codes.wmo.int/common/unit/", "@vocab": "https://api.weather.gov/ontology#", "geometry": { "@id": "s:GeoCoordinates", "@type": "geo:wktLiteral" }, "city": "s:addressLocality", "state": "s:addressRegion", "distance": { "@id": "s:Distance", "@type": "s:QuantitativeValue" }, "bearing": { "@type": "s:QuantitativeValue" }, "value": { "@id": "s:value" }, "unitCode": { "@id": "s:unitCode", "@type": "@id" }, "forecastOffice": { "@type": "@id" }, "forecastGridData": { "@type": "@id" }, "publicZone": { "@type": "@id" }, "county": { "@type": "@id" } } ], "id": "https://api.weather.gov/points/40.7143,-74.006", "type": "Feature", "geometry": { "type": "Point", "coordinates": [ -74.006, 40.714300000000001 ] }, "properties": { "@id": "https://api.weather.gov/points/40.7143,-74.006", "@type": "wx:Point", "cwa": "OKX", "forecastOffice": "https://api.weather.gov/offices/OKX", "gridId": "OKX", "gridX": 33, "gridY": 35, "forecast": "https://api.weather.gov/gridpoints/OKX/33,35/forecast", "forecastHourly": "https://api.weather.gov/gridpoints/OKX/33,35/forecast/hourly", "forecastGridData": "https://api.weather.gov/gridpoints/OKX/33,35", "observationStations": "https://api.weather.gov/gridpoints/OKX/33,35/stations", "relativeLocation": { "type": "Feature", "geometry": { "type": "Point", "coordinates": [ -74.0279259, 40.745251000000003 ] }, "properties": { "city": "Hoboken", "state": "NJ", "distance": { "unitCode": "wmoUnit:m", "value": 3906.1522008034999 }, "bearing": { "unitCode": "wmoUnit:degree_(angle)", "value": 151 } } }, "forecastZone": "https://api.weather.gov/zones/forecast/NYZ072", "county": "https://api.weather.gov/zones/county/NYC061", "fireWeatherZone": "https://api.weather.gov/zones/fire/NYZ212", "timeZone": "America/New_York", "radarStation": "KDIX" } } We use EvaluateJSONPath to grab a forecast URL. Then we call that forecast URL via invokeHTTP. That produces a larger JSON output that we will parse for the results we want to return to Slack. { "@context": [ "https://geojson.org/geojson-ld/geojson-context.jsonld", { "@version": "1.1", "wx": "https://api.weather.gov/ontology#", "geo": "http://www.opengis.net/ont/geosparql#", "unit": "http://codes.wmo.int/common/unit/", "@vocab": "https://api.weather.gov/ontology#" } ], "type": "Feature", "geometry": { "type": "Polygon", "coordinates": [ [ [ -74.025095199999996, 40.727052399999998 ], [ -74.0295579, 40.705361699999997 ], [ -74.000948300000005, 40.701977499999998 ], [ -73.996479800000003, 40.723667899999995 ], [ -74.025095199999996, 40.727052399999998 ] ] ] }, "properties": { "updated": "2023-11-15T14:34:46+00:00", "units": "us", "forecastGenerator": "BaselineForecastGenerator", "generatedAt": "2023-11-15T15:11:39+00:00", "updateTime": "2023-11-15T14:34:46+00:00", "validTimes": "2023-11-15T08:00:00+00:00/P7DT17H", "elevation": { "unitCode": "wmoUnit:m", "value": 2.1335999999999999 }, "periods": [ { "number": 1, "name": "Today", "startTime": "2023-11-15T10:00:00-05:00", "endTime": "2023-11-15T18:00:00-05:00", "isDaytime": true, "temperature": 51, "temperatureUnit": "F", "temperatureTrend": null, "probabilityOfPrecipitation": { "unitCode": "wmoUnit:percent", "value": null }, "dewpoint": { "unitCode": "wmoUnit:degC", "value": 2.2222222222222223 }, "relativeHumidity": { "unitCode": "wmoUnit:percent", "value": 68 }, "windSpeed": "1 to 7 mph", "windDirection": "SW", "icon": "https://api.weather.gov/icons/land/day/bkn?size=medium", "shortForecast": "Partly Sunny", "detailedForecast": "Partly sunny, with a high near 51. Southwest wind 1 to 7 mph." }, { "number": 2, "name": "Tonight", "startTime": "2023-11-15T18:00:00-05:00", "endTime": "2023-11-16T06:00:00-05:00", "isDaytime": false, "temperature": 44, "temperatureUnit": "F", "temperatureTrend": null, "probabilityOfPrecipitation": { "unitCode": "wmoUnit:percent", "value": null }, "dewpoint": { "unitCode": "wmoUnit:degC", "value": 3.8888888888888888 }, "relativeHumidity": { "unitCode": "wmoUnit:percent", "value": 82 }, "windSpeed": "8 mph", "windDirection": "SW", "icon": "https://api.weather.gov/icons/land/night/sct?size=medium", "shortForecast": "Partly Cloudy", "detailedForecast": "Partly cloudy, with a low around 44. Southwest wind around 8 mph." }, { "number": 3, "name": "Thursday", "startTime": "2023-11-16T06:00:00-05:00", "endTime": "2023-11-16T18:00:00-05:00", "isDaytime": true, "temperature": 60, "temperatureUnit": "F", "temperatureTrend": "falling", "probabilityOfPrecipitation": { "unitCode": "wmoUnit:percent", "value": null }, "dewpoint": { "unitCode": "wmoUnit:degC", "value": 5.5555555555555554 }, "relativeHumidity": { "unitCode": "wmoUnit:percent", "value": 82 }, "windSpeed": "6 mph", "windDirection": "SW", "icon": "https://api.weather.gov/icons/land/day/few?size=medium", "shortForecast": "Sunny", "detailedForecast": "Sunny. High near 60, with temperatures falling to around 58 in the afternoon. Southwest wind around 6 mph." }, { "number": 4, "name": "Thursday Night", "startTime": "2023-11-16T18:00:00-05:00", "endTime": "2023-11-17T06:00:00-05:00", "isDaytime": false, "temperature": 47, "temperatureUnit": "F", "temperatureTrend": null, "probabilityOfPrecipitation": { "unitCode": "wmoUnit:percent", "value": null }, "dewpoint": { "unitCode": "wmoUnit:degC", "value": 6.1111111111111107 }, "relativeHumidity": { "unitCode": "wmoUnit:percent", "value": 80 }, "windSpeed": "3 mph", "windDirection": "SW", "icon": "https://api.weather.gov/icons/land/night/few?size=medium", "shortForecast": "Mostly Clear", "detailedForecast": "Mostly clear, with a low around 47. Southwest wind around 3 mph." }, { "number": 5, "name": "Friday", "startTime": "2023-11-17T06:00:00-05:00", "endTime": "2023-11-17T18:00:00-05:00", "isDaytime": true, "temperature": 63, "temperatureUnit": "F", "temperatureTrend": "falling", "probabilityOfPrecipitation": { "unitCode": "wmoUnit:percent", "value": 20 }, "dewpoint": { "unitCode": "wmoUnit:degC", "value": 12.222222222222221 }, "relativeHumidity": { "unitCode": "wmoUnit:percent", "value": 86 }, "windSpeed": "2 to 10 mph", "windDirection": "S", "icon": "https://api.weather.gov/icons/land/day/bkn/rain,20?size=medium", "shortForecast": "Partly Sunny then Slight Chance Light Rain", "detailedForecast": "A slight chance of rain after 1pm. Partly sunny. High near 63, with temperatures falling to around 61 in the afternoon. South wind 2 to 10 mph. Chance of precipitation is 20%." }, { "number": 6, "name": "Friday Night", "startTime": "2023-11-17T18:00:00-05:00", "endTime": "2023-11-18T06:00:00-05:00", "isDaytime": false, "temperature": 51, "temperatureUnit": "F", "temperatureTrend": null, "probabilityOfPrecipitation": { "unitCode": "wmoUnit:percent", "value": 70 }, "dewpoint": { "unitCode": "wmoUnit:degC", "value": 12.777777777777779 }, "relativeHumidity": { "unitCode": "wmoUnit:percent", "value": 100 }, "windSpeed": "6 to 10 mph", "windDirection": "SW", "icon": "https://api.weather.gov/icons/land/night/rain,60/rain,70?size=medium", "shortForecast": "Light Rain Likely", "detailedForecast": "Rain likely. Cloudy, with a low around 51. Chance of precipitation is 70%. New rainfall amounts between a quarter and half of an inch possible." }, { "number": 7, "name": "Saturday", "startTime": "2023-11-18T06:00:00-05:00", "endTime": "2023-11-18T18:00:00-05:00", "isDaytime": true, "temperature": 55, "temperatureUnit": "F", "temperatureTrend": null, "probabilityOfPrecipitation": { "unitCode": "wmoUnit:percent", "value": 70 }, "dewpoint": { "unitCode": "wmoUnit:degC", "value": 11.111111111111111 }, "relativeHumidity": { "unitCode": "wmoUnit:percent", "value": 100 }, "windSpeed": "8 to 18 mph", "windDirection": "NW", "icon": "https://api.weather.gov/icons/land/day/rain,70/rain,30?size=medium", "shortForecast": "Light Rain Likely", "detailedForecast": "Rain likely before 1pm. Partly sunny, with a high near 55. Chance of precipitation is 70%." }, { "number": 8, "name": "Saturday Night", "startTime": "2023-11-18T18:00:00-05:00", "endTime": "2023-11-19T06:00:00-05:00", "isDaytime": false, "temperature": 40, "temperatureUnit": "F", "temperatureTrend": null, "probabilityOfPrecipitation": { "unitCode": "wmoUnit:percent", "value": null }, "dewpoint": { "unitCode": "wmoUnit:degC", "value": 1.1111111111111112 }, "relativeHumidity": { "unitCode": "wmoUnit:percent", "value": 65 }, "windSpeed": "12 to 17 mph", "windDirection": "NW", "icon": "https://api.weather.gov/icons/land/night/few?size=medium", "shortForecast": "Mostly Clear", "detailedForecast": "Mostly clear, with a low around 40." }, { "number": 9, "name": "Sunday", "startTime": "2023-11-19T06:00:00-05:00", "endTime": "2023-11-19T18:00:00-05:00", "isDaytime": true, "temperature": 50, "temperatureUnit": "F", "temperatureTrend": null, "probabilityOfPrecipitation": { "unitCode": "wmoUnit:percent", "value": null }, "dewpoint": { "unitCode": "wmoUnit:degC", "value": -0.55555555555555558 }, "relativeHumidity": { "unitCode": "wmoUnit:percent", "value": 65 }, "windSpeed": "10 to 14 mph", "windDirection": "W", "icon": "https://api.weather.gov/icons/land/day/few?size=medium", "shortForecast": "Sunny", "detailedForecast": "Sunny, with a high near 50." }, { "number": 10, "name": "Sunday Night", "startTime": "2023-11-19T18:00:00-05:00", "endTime": "2023-11-20T06:00:00-05:00", "isDaytime": false, "temperature": 38, "temperatureUnit": "F", "temperatureTrend": null, "probabilityOfPrecipitation": { "unitCode": "wmoUnit:percent", "value": null }, "dewpoint": { "unitCode": "wmoUnit:degC", "value": -0.55555555555555558 }, "relativeHumidity": { "unitCode": "wmoUnit:percent", "value": 67 }, "windSpeed": "13 mph", "windDirection": "NW", "icon": "https://api.weather.gov/icons/land/night/few?size=medium", "shortForecast": "Mostly Clear", "detailedForecast": "Mostly clear, with a low around 38." }, { "number": 11, "name": "Monday", "startTime": "2023-11-20T06:00:00-05:00", "endTime": "2023-11-20T18:00:00-05:00", "isDaytime": true, "temperature": 46, "temperatureUnit": "F", "temperatureTrend": null, "probabilityOfPrecipitation": { "unitCode": "wmoUnit:percent", "value": null }, "dewpoint": { "unitCode": "wmoUnit:degC", "value": -1.6666666666666667 }, "relativeHumidity": { "unitCode": "wmoUnit:percent", "value": 70 }, "windSpeed": "13 mph", "windDirection": "NW", "icon": "https://api.weather.gov/icons/land/day/sct?size=medium", "shortForecast": "Mostly Sunny", "detailedForecast": "Mostly sunny, with a high near 46." }, { "number": 12, "name": "Monday Night", "startTime": "2023-11-20T18:00:00-05:00", "endTime": "2023-11-21T06:00:00-05:00", "isDaytime": false, "temperature": 38, "temperatureUnit": "F", "temperatureTrend": null, "probabilityOfPrecipitation": { "unitCode": "wmoUnit:percent", "value": null }, "dewpoint": { "unitCode": "wmoUnit:degC", "value": -1.1111111111111112 }, "relativeHumidity": { "unitCode": "wmoUnit:percent", "value": 70 }, "windSpeed": "10 mph", "windDirection": "N", "icon": "https://api.weather.gov/icons/land/night/sct?size=medium", "shortForecast": "Partly Cloudy", "detailedForecast": "Partly cloudy, with a low around 38." }, { "number": 13, "name": "Tuesday", "startTime": "2023-11-21T06:00:00-05:00", "endTime": "2023-11-21T18:00:00-05:00", "isDaytime": true, "temperature": 49, "temperatureUnit": "F", "temperatureTrend": null, "probabilityOfPrecipitation": { "unitCode": "wmoUnit:percent", "value": 30 }, "dewpoint": { "unitCode": "wmoUnit:degC", "value": 2.7777777777777777 }, "relativeHumidity": { "unitCode": "wmoUnit:percent", "value": 73 }, "windSpeed": "9 to 13 mph", "windDirection": "E", "icon": "https://api.weather.gov/icons/land/day/bkn/rain,30?size=medium", "shortForecast": "Partly Sunny then Chance Light Rain", "detailedForecast": "A chance of rain after 1pm. Partly sunny, with a high near 49. Chance of precipitation is 30%." }, { "number": 14, "name": "Tuesday Night", "startTime": "2023-11-21T18:00:00-05:00", "endTime": "2023-11-22T06:00:00-05:00", "isDaytime": false, "temperature": 46, "temperatureUnit": "F", "temperatureTrend": null, "probabilityOfPrecipitation": { "unitCode": "wmoUnit:percent", "value": 50 }, "dewpoint": { "unitCode": "wmoUnit:degC", "value": 7.7777777777777777 }, "relativeHumidity": { "unitCode": "wmoUnit:percent", "value": 86 }, "windSpeed": "13 to 18 mph", "windDirection": "S", "icon": "https://api.weather.gov/icons/land/night/rain,50?size=medium", "shortForecast": "Chance Light Rain", "detailedForecast": "A chance of rain. Mostly cloudy, with a low around 46. Chance of precipitation is 50%." } ] } } We parse the data with EvaluateJSONPath to get primary fields for the weather. We then format those fields to PutSlack. LLM Skipped. Read forecast on ${date} for ${weatherlocation} @ ${latitude},${longitude} Used ${forecasturl} ${icon} Temp: ${temperature} ${temperatureunit} - ${temperaturetrend} There is a wind ${winddirection} at ${windspeed}. ${detailedforecast} Slack Output If we do have an LLM question, let’s make sure it’s just one record. We use a few different models that are available at IBM WatsonX.AI on IBM Cloud to quickly be accessed by our REST prompts. I tested and built the prompts initially at IBM’s Prompt Lab and then copied the initial curl statement from there. Click here for supported foundation models available with IBM watsonx.ai. ibm/mpt-7b-instruct2meta-llama/llama-2–70b-chatibm/granite-13b-chat-v1 We have to send our unique secure key to IBM and they will give us a token to use in our next call. We parse out the question and then send it to WatsonX via REST API. We build a prompt to send to IBM as follows. { "model_id": "meta-llama/llama-2-70b-chat", "input": "${inputs:urlEncode()}", "parameters": { "decoding_method": "greedy", "max_new_tokens": 200, "min_new_tokens": 50, "stop_sequences": [], "repetition_penalty": 1 }, "project_id": "0ead8ec4-d137-4f9c-8956-50b0da4a7068" } We parse the generated text which is our Generative AI results plus some helpful metadata on timings. The result posted to Slack is as follows: “You can use Apache NiFi to integrate Generative AI models in several ways: Data Preprocessing: Use NiFi to preprocess data before feeding it into your Generative AI model. This can include data cleaning, transformation, and feature engineering. Model Training: Use NiFi to automate the training process of your Generative AI model. You can use NiFi’s PutFile and PutFile_SFTP processors to write the training data to a file, and then use a processor like ExecuteScript to run the training script. Model Deployment: Once your Generative AI model is trained, you can use NiFi to deploy it. You can create a NiFi flow that takes in input data, runs it through the Generative AI model, and then outputs the generated data. Real-time Inference: You can use NiFi’s StreamingJobs” After the Slackbot posted the results, it posted metrics and debugging information to the chat channel. All of the metadata is posted to another Slack channel for administrator monitoring. ==== NiFi to IBM WatsonX.AI LLM Answers On Date: Wed, 15 Nov 2023 15:43:29 GMT Created: 2023-11-15T15:43:29.248Z Prompt: Q: What is a good way to integrate Generative AI and Apache NiFi? Response: ) You can use Apache NiFi to integrate Generative AI models in several ways: 1. Data Preprocessing: Use NiFi to preprocess data before feeding it into your Generative AI model. This can include data cleaning, transformation, and feature engineering. 2. Model Training: Use NiFi to automate the training process of your Generative AI model. You can use NiFi's PutFile and PutFile_SFTP processors to write the training data to a file, and then use a processor like ExecuteScript to run the training script. 3. Model Deployment: Once your Generative AI model is trained, you can use NiFi to deploy it. You can create a NiFi flow that takes in input data, runs it through the Generative AI model, and then outputs the generated data. 4. Real-time Inference: You can use NiFi's StreamingJobs Token: 200 Req Duration: 8153 HTTP TX ID: 89d71099-da23-4e7e-89f9-4e8f5620c0fb IBM Msg: This model is a Non-IBM Product governed by a third-party license that may impose use restrictions and other obligations. By using this model you agree to its terms as identified in the following URL. URL: https://dataplatform.cloud.ibm.com/docs/content/wsj/analyze-data/fm-models.html?context=wx IBM Msg ID: disclaimer_warning Model ID: meta-llama/llama-2-70b-chat Stop Reason: max_tokens Token Count: 38 TX ID: NGp0djg-c05f740f84f84b7c80f93f9da05aa756 UUID: da0806cb-6133-4bf4-808e-1fbf419c09e3 Corr ID: NGp0djg-c05f740f84f84b7c80f93f9da05aa756 Global TX ID: 20c3a9cf276c38bcdaf26e3c27d0479b Service Time: 478 Request ID: 03c2726a-dcb6-407f-96f1-f83f20fe9c9c File Name: 1a3c4386-86d2-4969-805b-37649c16addb Request Duration: 8153 Request URL: https://us-south.ml.cloud.ibm.com/ml/v1-beta/generation/text?version=2023-05-29 cf-ray: 82689bfd28e48ce2-EWR ===== Make Your Own Slackbot Slack Output Kafka Distribute Apache Flink SQL Table Creation DDL CREATE TABLE `ssb`.`Meetups`.`watsonairesults` ( `date` VARCHAR(2147483647), `x_global_transaction_id` VARCHAR(2147483647), `x_request_id` VARCHAR(2147483647), `cf_ray` VARCHAR(2147483647), `inputs` VARCHAR(2147483647), `created_at` VARCHAR(2147483647), `stop_reason` VARCHAR(2147483647), `x_correlation_id` VARCHAR(2147483647), `x_proxy_upstream_service_time` VARCHAR(2147483647), `message_id` VARCHAR(2147483647), `model_id` VARCHAR(2147483647), `invokehttp_request_duration` VARCHAR(2147483647), `message` VARCHAR(2147483647), `uuid` VARCHAR(2147483647), `generated_text` VARCHAR(2147483647), `transaction_id` VARCHAR(2147483647), `tokencount` VARCHAR(2147483647), `generated_token` VARCHAR(2147483647), `ts` VARCHAR(2147483647), `advisoryId` VARCHAR(2147483647), `eventTimeStamp` TIMESTAMP(3) WITH LOCAL TIME ZONE METADATA FROM 'timestamp', WATERMARK FOR `eventTimeStamp` AS `eventTimeStamp` - INTERVAL '3' SECOND ) WITH ( 'deserialization.failure.policy' = 'ignore_and_log', 'properties.request.timeout.ms' = '120000', 'format' = 'json', 'properties.bootstrap.servers' = 'kafka:9092', 'connector' = 'kafka', 'properties.transaction.timeout.ms' = '900000', 'topic' = 'watsonxaillmanswers', 'scan.startup.mode' = 'group-offsets', 'properties.auto.offset.reset' = 'earliest', 'properties.group.id' = 'watsonxaillmconsumer' ) CREATE TABLE `ssb`.`Meetups`.`watsonxresults` ( `date` VARCHAR(2147483647), `x_global_transaction_id` VARCHAR(2147483647), `x_request_id` VARCHAR(2147483647), `cf_ray` VARCHAR(2147483647), `inputs` VARCHAR(2147483647), `created_at` VARCHAR(2147483647), `stop_reason` VARCHAR(2147483647), `x_correlation_id` VARCHAR(2147483647), `x_proxy_upstream_service_time` VARCHAR(2147483647), `message_id` VARCHAR(2147483647), `model_id` VARCHAR(2147483647), `invokehttp_request_duration` VARCHAR(2147483647), `message` VARCHAR(2147483647), `uuid` VARCHAR(2147483647), `generated_text` VARCHAR(2147483647), `transaction_id` VARCHAR(2147483647), `tokencount` VARCHAR(2147483647), `generated_token` VARCHAR(2147483647), `ts` VARCHAR(2147483647), `eventTimeStamp` TIMESTAMP(3) WITH LOCAL TIME ZONE METADATA FROM 'timestamp', WATERMARK FOR `eventTimeStamp` AS `eventTimeStamp` - INTERVAL '3' SECOND ) WITH ( 'deserialization.failure.policy' = 'ignore_and_log', 'properties.request.timeout.ms' = '120000', 'format' = 'json', 'properties.bootstrap.servers' = 'kafka:9092', 'connector' = 'kafka', 'properties.transaction.timeout.ms' = '900000', 'topic' = 'watsonxaillm', 'scan.startup.mode' = 'group-offsets', 'properties.auto.offset.reset' = 'earliest', 'properties.group.id' = 'allwatsonx1' ) Example Prompt {"inputs":"Please answer to the following question. What is the capital of the United States?"} IBM DB2 SQL alter table "DB2INST1"."TRAVELADVISORY" add column "summary" VARCHAR(2048); -- DB2INST1.TRAVELADVISORY definition CREATE TABLE "DB2INST1"."TRAVELADVISORY" ( "TITLE" VARCHAR(250 OCTETS) , "PUBDATE" VARCHAR(250 OCTETS) , "LINK" VARCHAR(250 OCTETS) , "GUID" VARCHAR(250 OCTETS) , "ADVISORYID" VARCHAR(250 OCTETS) , "DOMAIN" VARCHAR(250 OCTETS) , "CATEGORY" VARCHAR(4096 OCTETS) , "DESCRIPTION" VARCHAR(4096 OCTETS) , "UUID" VARCHAR(250 OCTETS) NOT NULL , "TS" BIGINT NOT NULL , "summary" VARCHAR(2048 OCTETS) ) IN "IBMDB2SAMPLEREL" ORGANIZE BY ROW; ALTER TABLE "DB2INST1"."TRAVELADVISORY" ADD PRIMARY KEY ("UUID") ENFORCED; GRANT CONTROL ON TABLE "DB2INST1"."TRAVELADVISORY" TO USER "DB2INST1"; GRANT CONTROL ON INDEX "SYSIBM "."SQL230620142604860" TO USER "DB2INST1"; SELECT "summary", TITLE , ADVISORYID , TS, PUBDATE FROM DB2INST1.TRAVELADVISORY t WHERE "summary" IS NOT NULL ORDER BY ts DESC Example Output Email GitHub README GitHub repo Video Source Code Source Code

By Tim Spann CORE