Databases

When it comes to managing large amounts of data in a distributed system, Apache Cassandra and Apache Pulsar are two names that often come up. Apache Cassandra is a highly scalable NoSQL database that excels at handling high-velocity writes and queries across multiple nodes. It is an ideal solution for use cases such as user profile management, product catalogs, and real-time analytics. A platform for distributed messaging and streaming, called Apache Pulsar, was created to manage moving data. It can handle standard messaging workloads and more complex streaming use cases including real-time data processing and event-driven architectures. This article covers the main steps of building a Spring Boot and React-based web application that interacts with Pulsar and Cassandra, displaying stock data live as it is received. This is not a complete tutorial, it only covers the most important steps. You can find the complete source code for the application on GitHub. You will learn how to: Set up Cassandra and Pulsar instances using DataStax Astra DB and Astra Streaming. Publish and consume Pulsar messages in a Spring Boot application. Store Pulsar messages in Cassandra using a sink. Viewing live and stored data in React using the Hilla framework by Vaadin. Used Technologies and Libraries Apache Cassandra (with Astra DB) Apache Pulsar (with Astra Streaming) Spring Boot Spring for Apache Pulsar Spring Data for Apache Cassandra React Hilla AlphaVantage API Requirements Java 17 or newer Node 18 or newer Intermediate Java skills and familiarity with Spring Boot Storing Sensitive Data in Spring Boot Much of the setup for Cassandra and Pulsar is configuration-based. While it might be tempting to put the configuration in application.properties, it is not a smart idea as the file is under source control, and you may unintentionally reveal secrets. Instead, create a local config/local/application.properties configuration file and add it to .gitignore to ensure it does not leave your computer. The settings from the local configuration file will be automatically applied by Spring Boot: mkdir -p config/local touch config/local/application.properties echo " # Contains secrets that shouldn't go into the repository config/local/" >> .gitignore You may provide Spring Boot with the options as environment variables when using it in production. Setting Up Cassandra and Pulsar Using DataStax Astra Both Apache technologies used in this article are open-source projects and can be installed locally. However, using cloud services to set up the instances is a simpler option. In this article, we set up the data infrastructure required for our example web application using DataStax free tier services. Begin by logging in to your existing account or signing up for a new one on Astra DataStax’s official website, where you will be required to create the database and streaming service separately. Cassandra Setup Start by clicking “Create Database” from the official Astra DataStax website. Sinking data from a stream into Astra DB requires that both services are deployed in a region that supports both Astra Streaming and Astra DB: Enter the name of your new database instance. Select the keyspace name. (A keyspace stores your group of tables, a bit like schema in relational databases). Select a cloud Provider and Region.Note: For the demo application to work, you need to deploy the database service on a region that supports streaming too. Select “Create Database.” Cassandra: Connecting to the Service Once the initialization of the database service is created, you need to generate a token and download the “Secure Connection Bundle” that encrypts the data transfer between the app and the cloud database (mTLS). Navigate to the DB dashboard “Connect” tab sheet where you will find the button to generate a one-time token (please remember to download it) and the bundle download button: spring.cassandra.schema-action=CREATE_IF_NOT_EXISTS spring.cassandra.keyspace-name=<KEYSPACE_NAME> spring.cassandra.username=<ASTRADB_TOKEN_CLIENT_ID> spring.cassandra.password=<ASTRADB_TOKEN_SECRET> # Increase timeouts when connecting to Astra from a dev workstation spring.cassandra.contact-points=<ASTRADB_DATACENTER_ID> spring.cassandra.port=9042 spring.cassandra.local-datacenter=<ASTRADB_REGION> datastax.astra.secure-connect-bundle=<secure-connect-astra-stock-db.zip> Pulsar parameters for application.properties. Pulsar Set Up Start by clicking “Create Stream” from the main Astra DataStax page: Enter the name for your new streaming instance. Select a provider and region.Note: Remember to use the same provider and region you used to create the database service. Select “Create Stream.” Pulsar: Enabling Auto Topic Creation In addition to getting the streaming service up and running, you will also need to define the topic that is used by the application to consume and produce messages. You can create a topic explicitly using UI, but a more convenient way is to enable “Allow Auto Topic Creation” setting for the created instance: Click on the newly created stream instance and navigate to the “Namespace and Topics” tab sheet, and click “Modify Namespace.” Navigate to the “Settings” tab located under the default namespace (not the top-level “Settings” tab) and scroll all the way down. Change the “Allow Topic Creation” to “Allow Auto Topic Creation.” Changing this default setting will allow the application to create new topics automatically without any additional admin effort in Astra. With this, you have successfully established the infrastructure for hosting your active and passive data. Pulsar: Connecting to the Service Once the streaming instance has been set up, you need to create a token to access the service from your app. Most of the necessary properties are located on the “Connect” tab sheet of the “Streaming dashboard.” The “topic-name” input is found in the “Namespaces and Topics” tab sheet: ## Client spring.pulsar.client.service-url=<Broker Service URL> spring.pulsar.client.auth-plugin-class-name=org.apache.pulsar.client.impl.auth.AuthenticationToken spring.pulsar.client.authentication.token=<Astra_Streaming_Token> ## Producer spring.pulsar.producer.topic-name=persistent://<TENANT_NAME>/default/<TOPIC_NAME> spring.pulsar.producer.producer-name=<name of your choice> ## Consumer spring.pulsar.consumer.topics=persistent://<TENANT_NAME>/default/<TOPIC_NAME> spring.pulsar.consumer.subscription-name=<name of your choice> spring.pulsar.consumer.consumer-name=<name of your choice> spring.pulsar.consumer.subscription-type=key_shared Pulsar parameters for application.properties. Publishing Pulsar Messages From Spring Boot The Spring for Apache Pulsar library takes care of setting up Pulsar producers and consumers based on the given configuration. In the application, the StockPriceProducer component handles message publishing. To fetch stock data, it makes use of an external API call before publishing it to a Pulsar stream using a PulsarTemplate. Autowire the PulsarTemplate into the class and save it to a field: Java @Component public class StockPriceProducer { private final PulsarTemplate<StockPrice> pulsarTemplate; public StockPriceProducer(PulsarTemplate<StockPrice> pulsarTemplate) { this.pulsarTemplate = pulsarTemplate; } //... } Then use it to publish messages: Java private void publishStockPrices(Stream<StockPrice> stockPrices) { // Publish items to Pulsar with 100ms intervals Flux.fromStream(stockPrices) // Delay elements for the demo, don't do this in real life .delayElements(Duration.ofMillis(100)) .subscribe(stockPrice -> { try { pulsarTemplate.sendAsync(stockPrice); } catch (PulsarClientException e) { throw new RuntimeException(e); } }); } You need to configure the schema for the custom StockPrice type. In Application.java, define the following bean: Java @Bean public SchemaResolver.SchemaResolverCustomizer<DefaultSchemaResolver> schemaResolverCustomizer() { return (schemaResolver) -> schemaResolver.addCustomSchemaMapping(StockPrice.class, Schema.JSON(StockPrice.class)); } Consuming Pulsar Messages in Spring Boot The Spring for Apache Pulsar library comes with a @PulsarListener annotation for a convenient way of listening to Pulsar messages. Here, the messages are emitted to a Project Reactor Sink so the UI can consume them as a Flux: Java @Service public class StockPriceConsumer { private final Sinks.Many<StockPrice> stockPriceSink = Sinks.many().multicast().directBestEffort(); private final Flux<StockPrice> stockPrices = stockPriceSink.asFlux(); @PulsarListener private void stockPriceReceived(StockPrice stockPrice) { stockPriceSink.tryEmitNext(stockPrice); } public Flux<StockPrice> getStockPrices() { return stockPrices; } } Creating a Server Endpoint for Accessing Data From React The project uses Hilla, a full-stack web framework for Spring Boot. It manages websocket connections for reactive data types and allows type-safe server communication. The client may utilize the matching TypeScript methods created by the StockPriceEndpoint to fetch data: Java @Endpoint @AnonymousAllowed public class StockPriceEndpoint { private final StockPriceProducer producer; private final StockPriceConsumer consumer; private final StockPriceService service; StockPriceEndpoint(StockPriceProducer producer, StockPriceConsumer consumer, StockPriceService service) { this.producer = producer; this.consumer = consumer; this.service = service; } public List<StockSymbol> getSymbols() { return StockSymbol.supportedSymbols(); } public void produceDataForTicker(String ticker) { producer.produceStockPriceData(ticker); } public Flux<StockPrice> getStockPriceStream() { return consumer.getStockPrices(); } public List<StockPrice> findAllByTicker(String ticker) { return service.findAllByTicker(ticker); } } Displaying a Live-Updating Chart in React The DashboardView has an Apex Chart candle stick chart for displaying the stock data. It’s bound to a state of type ApexAxisChartSeries: TypeScript const [series, setSeries] = useState<ApexAxisChartSeries>([]); The view uses a React effect hook to call the server endpoint and subscribe to new data. It returns a disposer function to close the websocket when it is no longer needed: TypeScript useEffect(() => { const subscription = StockPriceEndpoint .getStockPriceStream() .onNext((stockPrice) => updateSeries(stockPrice)); return () => subscription.cancel(); }, []); The series is bound to the template. Because the backend and frontend are reactive, the chart is automatically updated any time a new Pulsar message is received: HTML <ReactApexChart type="candlestick" options={options} series={series} height={350} ></div> Persisting Pulsar Messages to Cassandra Sinking Pulsar messages to Astra DB can be useful in scenarios where you need a reliable, scalable, and secure platform to store event data from Pulsar for further analysis, processing, or sharing. Perhaps you need to retain a copy of event data for compliance and auditing purposes, need to store event data from multiple tenants in a shared database, or for some other use case. Astra Streaming offers numerous fully-managed Apache Pulsar connectors you can use to persist event data to various databases and third party solutions, like Snowflake. In this article, we are persisting the stream data into Astra DB. Creating a Sink Start by selecting the “Sink” tab sheet from the Astra streaming dashboard. Select the “default” namespace: From the list of available “Sink Types,” choose “Astra DB.” Give the sink a name of your choice Select the “stock-feed” that will be available once you have published messages to that topic from your app. After selecting data stream input, select the database you want to persist pulsar messages: To enable table creation, paste the Astra DB token with valid roles. You’ll notice keyspaces after the entry of a valid token, choose the keyspace name that was used to create the database. Then enter the table name.Note: This needs to match the @Table("stock_price") annotation value you use in StockPrice.java class to read back the data. Next, you need to map the properties from the Pulsar message to the database table column. Property fields are automatically mapped in our demo application, so you can simply click “Create” to proceed. If you were, for instance, persisting a portion of the data to the database, opening the schema definition would enable you to view the property names employed and create a custom mapping between the fields. After the sink is created, the initialization process will begin. After which, the status will change to “active.” Then, you’re done with automatically persisting stock data into your database for easy access by application. The sink dashboard provides access to sink log files in the event of an error. Displaying Cassandra Data in a Table The historical data that is stored in Cassandra are displayed in a data grid component. The DetailsView contains a Vaadin Grid component that is bound to an array of StockPrice objects which are kept in a state variable: TypeScript const [stockData, setStockData] = useState<StockPrice[]>([]); The view has a dropdown selector for selecting the stock you want to view. When the selection is updated, the view fetches the data for that stock from the server endpoint: TypeScript async function getDataFor(ticker?: string) { if (ticker) setStockData(await StockPriceEndpoint.findAllByTicker(ticker)); } The StockData array is bound to the grid in the template. GridColumns define the properties that columns should map to: HTML <Grid items={stockData} className="flex-grow"> <GridColumn path="time" ></GridColumn> <GridColumn path="open" ></GridColumn> <GridColumn path="high" ></GridColumn> <GridColumn path="low" ></GridColumn> <GridColumn path="close" ></GridColumn> <GridColumn path="volume" ></GridColumn> </Grid> Conclusion In this article, we showed how you can build a scalable real-time application using an open-source Java stack. You can clone the completed application and use it as a base for your own experiments.

In a contemporary application design that utilizes microservices and keeps up with dynamic analytics demands, it’s common to synchronize data from various databases dispersed throughout an enterprise and integrated with diverse systems. In modern applications, real-time syncing of databases is often necessary, so the synchronization cannot wait for a batch job to run daily or even hourly. The “Change Data Capture” concept addresses this need for real-time syncing by capturing and tracking any changes made to the data in the source databases and propagating those changes to the target databases in real time. As organizations continue to generate and store large amounts of data, the need for efficient and reliable real-time data replication has become increasingly important. Two solutions in this space are Debezium and DBConvert Streams. Both platforms promise to replicate data between different types of databases, but which one is right for you? In this article, we’ll compare DBConvert Streams and Debezium in terms of performance. Debezium/Kafka Architecture Source: Debezium The architecture for “Change Data Capture” using Debezium and Kafka Connect involves several components working together to capture, process, and route the data changes: Source database: The source database is where the changes occur, emitting events on inserts, updates, and deletes. Debezium connector: Debezium provides connectors for different databases, which capture the changes in a database-agnostic way and convert them into a standard message format. Apache Kafka: Kafka is a distributed streaming platform that acts as a message queue and ensures the messages are delivered in order and without loss. Kafka Connect: Kafka Connect is a component of Kafka that handles the integration between Kafka and external systems. It is used to pull data from Debezium and push it to the destination systems. Consumer applications: These applications consume the captured data changes and process them according to the use case. They could be analytics applications, data warehouses, or any other application that requires real-time data. DBConvert Streams Architecture In the DBConvert Streams architecture, the flow of data is as follows: The DBConvert Streams database source reader component reads CDC changes from the upstream source database. The source reader component propagates the changes to the NATS message broker. DBConvert Streams (DBS) uses NATS instead of Kafka in its architecture for Change Data Capture. The NATS message broker acts as a message queue, ensuring reliable and efficient delivery. The DBS target writer component consumes the events from the NATS message broker and writes them to the target database. This architecture enables real-time data integration between different databases, making it easier for development teams to work with diverse databases and reducing the amount of custom code needed to implement CDC. Different Database Log Formats There is no common standard for the format of database logs. Each database system typically uses its proprietary format for its transaction logs. This lack of standardization can make it challenging for developers to work with change data capture, particularly in multi-database environments. To address this issue, Debezium and DBConvert Streams offer connectors for popular databases, thereby simplifying the implementation of CDC. Performance Tests We will set up the test environment using Docker Compose to start all services and conduct performance tests to compare the efficiency of Debezium and DBConvert Streams in replicating one million records from a MySQL source database to a Postgres target database. We have created a GitHub repository to store the test scripts and results: Debezium tests DBConvert Streams tests Prerequisites Before proceeding, please ensure you have installed the necessary prerequisites, including: Docker Docker Compose Curl Please clone the specified repository onto your local machine: git clone git@github.com:slotix/dbconvert-streams-public.git Table Structure Source and Target Databases mysql-source database image is based on slotix/dbs-mysql:8, which has all the necessary settings to enable MySQL CDC replication. This image also contains the initdb.sql script, which creates a table with the above structure. postgres-target database is based on the official postgres:15-alpine image. It will receive all changes made to the mysql-source database. These databases are typically hosted on separate physical servers in a production environment. However, we will run them on a single machine using separate containers for our example. Debezium Test cd dbconvert-streams-public/examples/mysql2postgres/1- million-records-debezium export DEBEZIUM_VERSION=2.0 docker-compose up --build -d Set the environment variable DEBEZIUM_VERSION to 2.0 and then run the docker-compose up command with the options --build to build the images and -d to start the containers in the background. Deployment curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" http://localhost:8083/connectors/ -d @source.json curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" http://localhost:8083/connectors/ -d @target.json These two curl commands create Kafka Connect connectors for Debezium to read data from the MySQL source database and write data to the Postgres target database. Visit this link to view a list of active and running connectors. Monitor the Number of Records on the Target To ensure the replication process is working correctly, we will monitor the number of records on the target database using the script below: #!/bin/bash count=0 start_time=0 while true; do output=$(docker compose exec postgres-target bash -c "psql -U \$POSTGRES_USER \$POSTGRES_DB -c 'select count(*) from products'") count=$(echo "$output" | sed -n '3p' | awk '{print $1}') if [[ $count -gt 1 && $start_time -eq 0 ]]; then start_time=$(date +%s) echo "Started at" fi if [[ $count -gt 1000000 ]]; then end_time=$(date +%s) elapsed_time=$((end_time - start_time)) echo "Record count exceeded 1000000 after $elapsed_time seconds." break fi echo "$output" | sed -n '3p' | awk '{print $1}' sleep 1 done The script will print the number of records in the target database every second. Once the number of records exceeds 1 million, the script will print the elapsed time and exit. In the next terminal run the following commands: export DEBEZIUM_VERSION=2.0 ./count.sh Populate the Source Table With Sample Data To execute the SQL script that populates the source table with sample data, you can run the following commands in the first terminal: docker compose exec -it \ mysql-source \ mysql -u root -p123456 -D source In the MySQL prompt, execute the following command: INSERT INTO products (name, price, weight) SELECT CONCAT('Product', number) AS name, ROUND(RAND() * 100, 2) AS price, RAND() * 10 AS weight FROM (SELECT @row := @row + 1 AS number FROM (SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION SELECT 5 UNION SELECT 6 UNION SELECT 7 UNION SELECT 8 UNION SELECT 9 UNION SELECT 10) t1, (SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION SELECT 5 UNION SELECT 6 UNION SELECT 7 UNION SELECT 8 UNION SELECT 9 UNION SELECT 10) t2, (SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION SELECT 5 UNION SELECT 6 UNION SELECT 7 UNION SELECT 8 UNION SELECT 9 UNION SELECT 10) t3, (SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION SELECT 5 UNION SELECT 6 UNION SELECT 7 UNION SELECT 8 UNION SELECT 9 UNION SELECT 10) t4, (SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION SELECT 5 UNION SELECT 6 UNION SELECT 7 UNION SELECT 8 UNION SELECT 9 UNION SELECT 10) t5, (SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION SELECT 5 UNION SELECT 6 UNION SELECT 7 UNION SELECT 8 UNION SELECT 9 UNION SELECT 10) t6, (SELECT @row := 0) r ) numbers LIMIT 1000000; This command will insert 1 million rows of random data into a table named products. Results Once the operation is completed, you will see something like this in the terminal where the count.sh script is running: .... 942882 960702 977532 995352 Record count exceeded 1000000 after 96 seconds. That means it took about 96 seconds to replicate 1 million records from MySQL to Postgres Database. DBConvert Streams Test We are going to make the same test with DBConvert Streams: cd dbconvert-streams-public/examples/mysql2postgres/1-million-records docker-compose up --build -d The command above will start the services in the background, build the images, and use the docker-compose.yml file to configure the services. Send Stream Configuration Send a request to the DBConvert Streams API with configuration parameters. Run the curl command: curl --request POST --url http://127.0.0.1:8020/api/v1/streams\?file=./mysql2pg.json Populate the Source Table With Sample Data Connect to the MySQL container to interact with the MySQL database running inside the container: docker exec -it \ mysql-source \ mysql -uroot -p123456 -D source In the MySQL prompt, execute the following command: INSERT INTO products (name, price, weight) SELECT CONCAT('Product', number) AS name, ROUND(RAND() * 100, 2) AS price, RAND() * 10 AS weight FROM (SELECT @row := @row + 1 AS number FROM (SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION SELECT 5 UNION SELECT 6 UNION SELECT 7 UNION SELECT 8 UNION SELECT 9 UNION SELECT 10) t1, (SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION SELECT 5 UNION SELECT 6 UNION SELECT 7 UNION SELECT 8 UNION SELECT 9 UNION SELECT 10) t2, (SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION SELECT 5 UNION SELECT 6 UNION SELECT 7 UNION SELECT 8 UNION SELECT 9 UNION SELECT 10) t3, (SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION SELECT 5 UNION SELECT 6 UNION SELECT 7 UNION SELECT 8 UNION SELECT 9 UNION SELECT 10) t4, (SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION SELECT 5 UNION SELECT 6 UNION SELECT 7 UNION SELECT 8 UNION SELECT 9 UNION SELECT 10) t5, (SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION SELECT 5 UNION SELECT 6 UNION SELECT 7 UNION SELECT 8 UNION SELECT 9 UNION SELECT 10) t6, (SELECT @row := 0) r ) numbers LIMIT 1000000; This command will insert 1 million rows of random data into a table named products. Results Copy and paste the URL into your web browser to access the Prometheus metrics page: http://127.0.0.1:9090/graph?g0.expr=source_events_in&g0.tab=1&g0.stacked=0&g0.show_exemplars=0&g0.range_input=1h&g1.expr=source_duration_seconds&g1.tab=1&g1.stacked=0&g1.show_exemplars=0&g1.range_input=1h&g2.expr=target_events_out&g2.tab=1&g2.stacked=0&g2.show_exemplars=0&g2.range_input=1h&g3.expr=target_duration_seconds&g3.tab=1&g3.stacked=0&g3.show_exemplars=0&g3.range_input=1h Once you’re on the page, you’ll be able to see various performance metrics related to the replication process, including the number of events in and out of the source and target databases and the duration of the replication process. It took about 8 seconds to transfer 1 million records from MySQL to Postgres. Minimum Hardware Requirements We conducted additional tests as described above on the DigitalOcean cloud computing platform to find out the information about the minimum hardware resources Debezium and DBConvert Streams requires. The table below summarizes the test results. The table below summarizes the test results of replication of 1 Million records from MySQL to PostgreSQL: HARDWARE RESOURCES DEBEZIUM DBCONVERT STREAMS 2 CPU / 2 GB RAM Failed 15 seconds 2 CPU / 4 GB RAM Failed (after ~300k records) 12 seconds 4 CPU / 8 GB RAM 236 seconds 8 seconds 8 CPU / 16 GB RAM 221 seconds 8 seconds According to the table, the DBConvert Streams platform requires fewer resources and provides faster replication speeds compared to Debezium. Conclusion While Debezium is a popular CDC platform that offers a lot of flexibility and powerful features, it may not always be the most cost-effective solution for some use cases. Specifically, DBConvert Streams may provide better performance and cost-efficiency for use cases requiring high throughput and real-time replication.

ClickHouse is an open-source real-time analytics database built and optimized for use cases requiring super-low latency analytical queries over large amounts of data. To achieve the best possible performance for analytical applications, combining tables in a data denormalization process is typical. Flattened tables help minimize query latency by avoiding joins at the cost of incremental ETL complexity, typically acceptable in return for sub-second queries. However, denormalizing data isn't always practical for some workloads, for instance, those coming from more traditional data warehouses. Sometimes, part of the source data for analytical queries needs to remain normalized. These normalized tables take less storage and provide flexibility with data combinations, but they require joins at query time for certain types of analysis. Fortunately, contrary to some misconceptions, joins are fully supported in ClickHouse! In addition to supporting all standard SQL JOIN types, ClickHouse provides additional JOIN types useful for analytical workloads and time-series analysis. ClickHouse allows you to choose between six different algorithms for the join execution or allow the query planner to adaptively choose and dynamically change the algorithm at runtime, depending on resource availability and usage. Join Types Supported in Clickhouse We use Venn diagrams and example queries on a normalized IMDB dataset originating from the relational dataset repository to explain the available join types in ClickHouse. Instructions for creating and loading the tables are here. The dataset is also available in our playground for users wanting to reproduce queries. We are going to use four tables from our example dataset: The data in that four tables represent movies. A movie can have one or many genres. The roles in a movie are played by actors. The arrows in the diagram above represent foreign-to-primary-key-relationships. e.g., the movie_idcolumn of a row in the genres table contains theid value from a row in the movies table. There is a many-to-many relationship between movies and actors. This many-to-many relationship is normalized into two one-to-many relationships by using the roles table. Each row in the roles table contains the values of the id fields of the movies table and the actors' table. Inner Join The Inner Join returns, for each pair of rows matching on join keys, the column values of the row from the left table, combined with the column values of the row from the right table. If a row has more than one match, then all matches are returned (meaning that the cartesian product is produced for rows with matching join keys). This query finds the genre(s) for each movie by joining the movies table with the genres table: SQL SELECT m.name AS name, g.genre AS genre FROM movies AS m INNER JOIN genres AS g ON m.id = g.movie_id ORDER BY m.year DESC, m.name ASC, g.genre ASC LIMIT 10; ┌─name───────────────────────────────────┬─genre─────┐ │ Harry Potter and the Half-Blood Prince │ Action │ │ Harry Potter and the Half-Blood Prince │ Adventure │ │ Harry Potter and the Half-Blood Prince │ Family │ │ Harry Potter and the Half-Blood Prince │ Fantasy │ │ Harry Potter and the Half-Blood Prince │ Thriller │ │ DragonBall Z │ Action │ │ DragonBall Z │ Adventure │ │ DragonBall Z │ Comedy │ │ DragonBall Z │ Fantasy │ │ DragonBall Z │ Sci-Fi │ └────────────────────────────────────────┴───────────┘ 10 rows in set. Elapsed: 0.126 sec. Processed 783.39 thousand rows, 21.50 MB (6.24 million rows/s., 171.26 MB/s.) Note that the Inner keyword can be omitted. The behavior of the Inner Join can be extended or changed by using one of the following other join types. (Left/Right/Full) Outer Join The Left Outer Join behaves like Inner Join; plus, for non-matching left table rows, ClickHouse returns default values for the right table’s columns. A Right Outer Join query is similar and also returns values from non-matching rows from the right table together with default values for the columns of the left table. A Full Outer Join query combines the left and right outer join and returns values from non-matching rows from the left and the right table, together with default values for the columns of the right and left table, respectively. Note that ClickHouse can be configured to return NULLs instead of default values (however, for performance reasons, that is less recommended). This query finds all movies that have no genre by querying for all rows from the movies table that don’t have matches in the genres table and therefore gets (at query time) the default value 0 for the movie_id column: SQL SELECT m.name FROM movies AS m LEFT JOIN genres AS g ON m.id = g.movie_id WHERE g.movie_id = 0 ORDER BY m.year DESC, m.name ASC LIMIT 10; ┌─name──────────────────────────────────────┐ │ """Pacific War, The""" │ │ """Turin 2006: XX Olympic Winter Games""" │ │ Arthur, the Movie │ │ Bridge to Terabithia │ │ Mars in Aries │ │ Master of Space and Time │ │ Ninth Life of Louis Drax, The │ │ Paradox │ │ Ratatouille │ │ """American Dad""" │ └───────────────────────────────────────────┘ 10 rows in set. Elapsed: 0.092 sec. Processed 783.39 thousand rows, 15.42 MB (8.49 million rows/s., 167.10 MB/s.) Note that the Outer keyword can be omitted. Cross Join The Cross Join produces the full cartesian product of the two tables without considering join keys. Each row from the left table is combined with each row from the right table. The following query, therefore, is combing each row from the movies table with each row from the genres table: SQL SELECT m.name, m.id, g.movie_id, g.genre FROM movies AS m CROSS JOIN genres AS g LIMIT 10; ┌─name─┬─id─┬─movie_id─┬─genre───────┐ │ #28 │ 0 │ 1 │ Documentary │ │ #28 │ 0 │ 1 │ Short │ │ #28 │ 0 │ 2 │ Comedy │ │ #28 │ 0 │ 2 │ Crime │ │ #28 │ 0 │ 5 │ Western │ │ #28 │ 0 │ 6 │ Comedy │ │ #28 │ 0 │ 6 │ Family │ │ #28 │ 0 │ 8 │ Animation │ │ #28 │ 0 │ 8 │ Comedy │ │ #28 │ 0 │ 8 │ Short │ └──────┴────┴──────────┴─────────────┘ 10 rows in set. Elapsed: 0.024 sec. Processed 477.04 thousand rows, 10.22 MB (20.13 million rows/s., 431.36 MB/s.) While the previous example query alone didn’t make much sense, it can be extended with a where clause for associating matching rows to replicate Inner join behavior for finding the genre(s) for each movie: SQL SELECT m.name, g.genre FROM movies AS m CROSS JOIN genres AS g WHERE m.id = g.movie_id ORDER BY m.year DESC, m.name ASC LIMIT 10; ┌─name───────────────────────────────────┬─genre─────┐ │ Harry Potter and the Half-Blood Prince │ Action │ │ Harry Potter and the Half-Blood Prince │ Adventure │ │ Harry Potter and the Half-Blood Prince │ Family │ │ Harry Potter and the Half-Blood Prince │ Fantasy │ │ Harry Potter and the Half-Blood Prince │ Thriller │ │ DragonBall Z │ Action │ │ DragonBall Z │ Sci-Fi │ │ DragonBall Z │ Fantasy │ │ DragonBall Z │ Comedy │ │ DragonBall Z │ Adventure │ └────────────────────────────────────────┴───────────┘ 10 rows in set. Elapsed: 0.441 sec. Processed 783.39 thousand rows, 21.50 MB (1.78 million rows/s., 48.78 MB/s.) An alternative syntax for Cross Join specifies multiple tables in the from clause separated by commas. ClickHouse is rewriting a Cross Join to an Inner Join if there are joining expressions in the where section of the query. We can check that for the example query via EXPLAIN SYNTAX (that returns the syntactically optimized version into which a query gets rewritten before being executed): SQL EXPLAIN SYNTAX SELECT m.name AS name, g.genre AS genre FROM movies AS m CROSS JOIN genres AS g WHERE m.id = g.movie_id ORDER BY m.year DESC, m.name ASC, g.genre ASC LIMIT 10; ┌─explain─────────────────────────────────────┐ │ SELECT │ │ name AS name, │ │ genre AS genre │ │ FROM movies AS m │ │ ALL INNER JOIN genres AS g ON id = movie_id │ │ WHERE id = movie_id │ │ ORDER BY │ │ year DESC, │ │ name ASC, │ │ genre ASC │ │ LIMIT 10 │ └─────────────────────────────────────────────┘ 11 rows in set. Elapsed: 0.077 sec. The Inner Join clause in the syntactically optimized Cross Join query version contains the all keyword, that got explicitly added in order to keep the cartesian product semantics of the Cross Join even when being rewritten into an Inner Join, for which the cartesian product can be disabled. And because, as mentioned above, the Outer keyword can be omitted for a Right Outer Join, and the optional all keyword can be added. You can write All Right Join and it will work all right. Left/Right Semi Join A Left Semi Join query returns column values for each row from the left table that has at least one join key match in the right table. Only the first found match is returned (the cartesian product is disabled). A Right Semi Join query is similar and returns values for all rows from the right table with at least one match in the left table, but only the first found match is returned. This query finds all actors/actresses that performed in a movie in 2023. Note that with a normal (Inner) join, the same actor/actress would show up more than one time if they had more than one role in 2023: SQL SELECT a.first_name, a.last_name FROM actors AS a LEFT SEMI JOIN roles AS r ON a.id = r.actor_id WHERE toYear(created_at) = '2023' ORDER BY id ASC LIMIT 10; ┌─first_name─┬─last_name──────────────┐ │ Michael │ 'babeepower' Viera │ │ Eloy │ 'Chincheta' │ │ Dieguito │ 'El Cigala' │ │ Antonio │ 'El de Chipiona' │ │ José │ 'El Francés' │ │ Félix │ 'El Gato' │ │ Marcial │ 'El Jalisco' │ │ José │ 'El Morito' │ │ Francisco │ 'El Niño de la Manola' │ │ Víctor │ 'El Payaso' │ └────────────┴────────────────────────┘ 10 rows in set. Elapsed: 0.151 sec. Processed 4.25 million rows, 56.23 MB (28.07 million rows/s., 371.48 MB/s.) Left/Right Anti Join A Left Anti Join returns column values for all non-matching rows from the left table. Similarly, the Right Anti Join returns column values for all non-matching right table rows. An alternative formulation of our previous outer join example query is using an anti join for finding movies that have no genre in the dataset: SQL SELECT m.name FROM movies AS m LEFT ANTI JOIN genres AS g ON m.id = g.movie_id ORDER BY year DESC, name ASC LIMIT 10; ┌─name──────────────────────────────────────┐ │ """Pacific War, The""" │ │ """Turin 2006: XX Olympic Winter Games""" │ │ Arthur, the Movie │ │ Bridge to Terabithia │ │ Mars in Aries │ │ Master of Space and Time │ │ Ninth Life of Louis Drax, The │ │ Paradox │ │ Ratatouille │ │ """American Dad""" │ └───────────────────────────────────────────┘ 10 rows in set. Elapsed: 0.077 sec. Processed 783.39 thousand rows, 15.42 MB (10.18 million rows/s., 200.47 MB/s.) Left/Right/Inner Any Join A Left Any Join is the combination of the Left Outer Join + the Left Semi Join, meaning that ClickHouse returns column values for each row from the left table, either combined with the column values of a matching row from the right table or combined with default column values for the right table, in case no match exists. If a row from the left table has more than one match in the right table, ClickHouse only returns the combined column values from the first found match (the cartesian product is disabled). Similarly, the Right Any Join is the combination of the Right Outer Join + the Right Semi Join. And the Inner Any Join is the Inner Join with a disabled cartesian product. We demonstrate the Left Any Join with an abstract example using two temporary tables (left_table and right_table) constructed with the values table function: SQL WITH left_table AS (SELECT * FROM VALUES('c UInt32', 1, 2, 3)), right_table AS (SELECT * FROM VALUES('c UInt32', 2, 2, 3, 3, 4)) SELECT l.c AS l_c, r.c AS r_c FROM left_table AS l LEFT ANY JOIN right_table AS r ON l.c = r.c; ┌─l_c─┬─r_c─┐ │ 1 │ 0 │ │ 2 │ 2 │ │ 3 │ 3 │ └─────┴─────┘ 3 rows in set. Elapsed: 0.002 sec. This is the same query using a Right Any Join: SQL WITH left_table AS (SELECT * FROM VALUES('c UInt32', 1, 2, 3)), right_table AS (SELECT * FROM VALUES('c UInt32', 2, 2, 3, 3, 4)) SELECT l.c AS l_c, r.c AS r_c FROM left_table AS l RIGHT ANY JOIN right_table AS r ON l.c = r.c; ┌─l_c─┬─r_c─┐ │ 2 │ 2 │ │ 2 │ 2 │ │ 3 │ 3 │ │ 3 │ 3 │ │ 0 │ 4 │ └─────┴─────┘ 5 rows in set. Elapsed: 0.002 sec. This is the query with an Inner Any Join: SQL WITH left_table AS (SELECT * FROM VALUES('c UInt32', 1, 2, 3)), right_table AS (SELECT * FROM VALUES('c UInt32', 2, 2, 3, 3, 4)) SELECT l.c AS l_c, r.c AS r_c FROM left_table AS l INNER ANY JOIN right_table AS r ON l.c = r.c; ┌─l_c─┬─r_c─┐ │ 2 │ 2 │ │ 3 │ 3 │ └─────┴─────┘ 2 rows in set. Elapsed: 0.002 sec. ASOF Join The ASOF Join, implemented for ClickHouse in 2019 by Martijn Bakker and Artem Zuikov, provides non-exact matching capabilities. If a row from the left table doesn’t have an exact match in the right table, then the closest matching row from the right table is used as a match instead. This is particularly useful for time-series analytics and can drastically reduce query complexity. We will do time-series analytics of stock market data as an example. A quotes table contains stock symbol quotes based on specific times of the day. The price is updated every 10 seconds in our example data. A trades table lists symbol trades - a specific volume of a symbol got bought at a specific time: In order to calculate the concrete cost of each trade, we need to match the trades with their closest quote time. This is easy and compact with the ASOF Join, where we use the ON clause for specifying an exact match condition and the AND clause for specifying the closest match condition — we are looking for the closest row from the quotes table exactly or before the date of a trade: SQL SELECT t.symbol, t.volume, t.time AS trade_time, q.time AS closest_quote_time, q.price AS quote_price, t.volume * q.price AS final_price FROM trades t ASOF LEFT JOIN quotes q ON t.symbol = q.symbol AND t.time >= q.time FORMAT Vertical; Row 1: ────── symbol: ABC volume: 200 trade_time: 2023-02-22 14:09:05 closest_quote_time: 2023-02-22 14:09:00 quote_price: 32.11 final_price: 6422 Row 2: ────── symbol: ABC volume: 300 trade_time: 2023-02-22 14:09:28 closest_quote_time: 2023-02-22 14:09:20 quote_price: 32.15 final_price: 9645 2 rows in set. Elapsed: 0.003 sec. Note that the ON clause of the ASOF Join is required and specifies an exact match condition next to the non-exact match condition of the AND clause. ClickHouse currently doesn't support (yet) joins without any part of the join keys performing strict matching. Summary This blog post showed how ClickHouse supports all standard SQL Join types, plus specialized joins to power analytical queries. We described and demonstrated all supported join types.

Recently, researchers at UC Berkeley published TAOBench, an end-to-end benchmark for social network workloads based on the traces collected from TAO, the distributed database for Meta. Among the typical data operations in social network applications, 99.7% are read, while only 0.2% are write and 0.01% are write transactions. Some other findings include: Transaction hotkeys are often co-located. Read and write hotspots appear on different keys. Contention can result intentionally. All these characteristics of social network application workloads pose special challenges. In this article, we will share the typical technological challenges faced by social network applications. Based on the findings from TAOBench, we will also discuss some features and optimizations of Distributed SQL databases that help optimize social network benchmark performance. The Underlying Technical Challenges of Social Networking Applications Social network applications allow users to share information, such as text, photos, videos, and links, and to interact with each other through various features, such as messaging, forums, and groups. These objects and features work together to describe relationships. As a user of a social application, you will have many needs that pose data challenges: You don't want unauthorized people to see your posts, but you do want people relevant to you to see them. You want your posts visible immediately, not delayed for hours. You don't want to miss replies that come in late. Late replies also confuse users. You want to access the information anytime, anywhere. The underlying technology challenges of social network applications include scalability, privacy, security, and data management. Scalability: This is the ability of a system to handle an increasing amount of traffic and data without becoming slow or unresponsive. Social network applications must handle large amounts of data and traffic: they often have millions of users who are constantly generating and accessing data. Privacy and security: These are vital for social network applications. These applications often contain sensitive personal information, such as user profiles, messages, and connections, and they need to protect this information from unauthorized access and disclosure. Data management: Applications need to be able to efficiently store, process, and retrieve large amounts of data while also ensuring data integrity and availability. All these needs are even more challenging at hyper-scale—a scale that increases with time. As the TAO paper mentions, "A single Facebook page may aggregate and filter hundreds of items from the social graph…. It must be efficient, highly available, and scale to high query rates." How Do Distributed SQL Databases Fit? To handle the high volume of data and traffic that social networks generate, database systems must be able to scale horizontally, using multiple servers and other resources to distribute the workload and improve performance. In addition to scalability, social network databases must be able to support fast and efficient querying to provide users with a smooth and responsive experience. This involves using specialized indexing and data structures, as well as complex query optimization algorithms. Distributed SQL databases are designed to handle large amounts of data and traffic and can be easily scaled horizontally across multiple servers and other resources. They may also offer features such as high availability, fault tolerance, and flexible data modeling, which can be useful for social network applications. Indications in the TAOBench Test In the TAOBench test, a UC Berkeley scholar tested several distributed cloud databases with equivalently priced infrastructure resources. The result is shown below. TAOBench’s benchmark on distributed databases The figure shows that the latency of all databases increases when the throughput reaches a certain level. (Each database has different scalability and performance limits.) This is because these databases (except for Cloud Spanner) are limited to resources of equal cost. Compared to other vendors, TiDB, a distributed SQL database, demonstrates stable performance with the best scalability. This allows it to achieve higher throughput. TiDB’s Architecture and Optimizations for Social Networking TiDB is a Distributed SQL database that offers scalability, high availability, ACID transactions, and MySQL compatibility, making it ideal for OLTP scenarios. Today, TiDB plays a vital role in many social networking companies, such as Zhihu (China's Quora), Redbook, Pinterest, and Bilibili. Many enterprises are using TiDB to help them deal with large-scale data issues. TiDB also provides true Hybrid Transactional/Analytical Processing (HTAP) capabilities that simplify technology stacks by combining OLTP and real-time analysis. As a Distributed SQL database, TiDB excels in TAOBench tests for performance and scaling. There are some good architectural reasons: Consistency and isolation: ACID and distributed transaction support based on Percolator High availability: data replicas based on Raft High throughput: horizontal elastic, scalable nodes to support Multi-write Relational data access: MySQL compatibility Ability to handle hotspot issues: Auto split and rebalance with the data region In addition, certain aspects of TiDB's design make it well-suited for networking applications. Auto Dynamic Sharding and Rebalancing As the TAOBench paper says, "Transaction hotkeys are often colocated." The hot spot issue is a difficult one in social networking applications. In TiDB, the fundamental data storage unit for management and distribution is called a "Region.” Regions can be split and merged based on the amount of data they manage and can be scheduled to move between nodes. Typically, data is distributed evenly across all storage nodes, and TiDB automatically balances the resources of each store based on the load. A storage node’s CPU and disk usage may become a bottleneck. TiDB’s Placement Driver (PD) estimates the load of the data regions based on the statistics such as the number of requests and amount of data written and synchronized. PD can schedule the balancing operation accordingly. Data region rebalancing with PD In a social network, hotspots may be concentrated within one data Region. TiDB samples a data region to analyze the distribution of workload. It then finds a suitable split point to allow the hot data Region to split into smaller Regions. After splitting, the hotspot balancing scheduler can move the hotspots into different nodes. With these two scheduling features, TiDB can fully utilize the distributed nature of storage, IO, and computing. This keeps performance stable — even in cases of severe hot-spotting. Hotspots processing in TiDB Write Transaction Optimization for Colocated Participants Distributed systems that support cross-row and cross-node transactions typically use 2-phase-commit (2PC) to achieve atomicity. TiDB's 2PC implementation is based on Percolator. In TiDB’s Percolator model, a transaction is considered committed once the primary key is committed. This requires at least two network round trips. However, not all transactions require 2PC to achieve atomicity. If a transaction only involves data hosted on one node, atomic commits can be achieved with only one round of RPCs. TiDB’s optimized process for write transactions The TAOBench paper says, "Hotkeys in write transactions tend to be colocated on the same shards." This optimization in TiDB effectively reduces the number of transaction commit phases. In the test results, we observed that Commit Operations per Second (OPS) dropped from 6,000 to less than 1,000, indicating that most 2PCs were reduced to 1PC. However, since writes in TAOBench only account for approximately 0.2% of all traffic, the overall Queries per Second (QPS) only saw a slight improvement. Commit performance observed in TAOBench A potential topic for future optimization is using data affinity to colocate as much relevant data as possible in one data region. This can reduce the overhead of 2PC and improve performance. Plan Cache for Read-Heavy Workload TiDB supports plan cache for SQL statements. This allows it to skip the optimization phase, which includes rule-based optimization and cost-based optimization. For read-heavy workloads, skipping these processes saves computing resources and improves performance. Based on our testing, enabling the plan cache improves QPS by about 10.5%. Plan cache in TiDB Semi-Auto-Tuning Garbage Collection For any data-intensive system, garbage collection (GC) is a resource-intensive background task. The GC threshold parameters can significantly affect the system's performance, particularly when it consumes a lot of CPU resources. Go GC auto-tuning, an optimization proposed by an Uber engineer, can reduce unnecessary GC operations and save overhead on frequent lightweight operations. TiDB adopted this optimization, which significantly improved throughput for TAOBench and many other Online Transactional Processing (OLTP) workloads. However, there’s a tradeoff. Although this method reduces unnecessarily frequent GC, in extreme cases, it may increase the risk of out-of-memory (OOM) crashes. Results can be found in the following graphic. Continuous Iteration and Evaluation In addition to the features and optimizations we’ve discussed, the evolution of the product itself is essential to address scalability challenges. TiDB iterates quickly and often, with a release cadence of one to two months. To capture the performance gains across different versions, the team also set up TAOBench to compare the performance of TiDB 6.4 with TiDB 5.0. As indicated in the chart, we achieved a 30% QPS improvement over the past year and a half. TiDB’s overall performance comparison on TAOBench Conclusion Overall, the workload of a social network database can be quite challenging. It requires advanced technologies and techniques to manage and optimize the data to give users a seamless and enjoyable experience. To learn more, you can watch this Meetup playback.

NoSQL stands for "Not Only SQL" and refers to a type of database management system that is designed to handle large volumes of unstructured and semi-structured data. Unlike traditional SQL databases that use a tabular format with predefined schemas, NoSQL databases are schema-less and allow for flexible and dynamic data structures. NoSQL databases are required because they can handle the large volumes and complex data types associated with Big Data. They are designed to scale horizontally by distributing data across many servers, making them well-suited for handling large and growing datasets. Additionally, NoSQL databases are often faster and more efficient than SQL databases for certain types of queries, such as those involving large amounts of data and complex data structures. NoSQL databases are also used in modern web applications that require fast and flexible data storage, such as social media platforms, online marketplaces, and content management systems. They are particularly useful for applications that require high levels of availability and scalability, as they can handle large amounts of traffic and data without sacrificing performance. Different Types of NoSQL Databases There are several types of NoSQL databases, each designed to handle different types of data and workloads. Some common types of NoSQL databases include: Document Databases These databases store and manage semi-structured data as documents, typically in JSON or XML formats. Document databases are well-suited for managing unstructured data, such as user profiles, product catalogs, or content management systems. Examples of document databases include MongoDB, Elasticsearch, and Couchbase. Key-Value Databases These databases store data as key-value pairs, making them ideal for simple lookups and high-speed data retrieval. Key-value databases are often used for caching, session management, and message queues. Examples of key-value databases include Redis and Riak. Column-Family Databases Also known as column-oriented databases, these databases store data as columns instead of rows, making them ideal for handling large amounts of data and complex queries. Column-family databases are often used for analytics, content management, and data warehousing. Examples of column-family databases include Apache Cassandra and HBase. Graph Databases These databases store and manage data as nodes and edges, making them well-suited for managing complex relationships and hierarchies. Graph databases are often used for social networks, recommendation engines, and fraud detection. Examples of graph databases include Neo4j and OrientDB. CAP Theorem for NoSQL Database The CAP theorem, also known as Brewer's theorem, is a fundamental concept in distributed computing that applies to NoSQL databases. The CAP theorem states that in any distributed system, it is impossible to simultaneously provide all three of the following guarantees: Consistency: Every read request from a node in the system will return the most recent write request. Availability: Every request to the system will receive a response without guaranteeing that it contains the most recent written request. Partition tolerance: The system can continue to operate and function correctly even if there are network partitions or messages are lost between nodes. In other words, when designing a distributed system like a NoSQL database, developers have to make trade-offs between consistency, availability, and partition tolerance. NoSQL databases are typically designed to prioritize either availability or partition tolerance while sacrificing some degree of consistency. This means that in certain failure scenarios, a NoSQL database may not provide the most up-to-date data to all nodes in the system but instead might return stale or conflicting data. For example, in a partitioned network, a NoSQL database may prioritize partition tolerance and continue to accept writes from multiple nodes, but these nodes may have different versions of the same data. In contrast, a traditional relational database might prioritize consistency and reject writes until it can guarantee that all nodes have the most recent data. Overall, the CAP theorem is an important consideration when designing and choosing a NoSQL database, as it helps to identify the trade-offs between consistency, availability, and partition tolerance that must be made in a distributed system. Use of NoSQL Database NoSQL databases are widely used for a variety of reasons, including: Scalability: NoSQL databases are highly scalable, allowing them to handle large amounts of data and high-traffic loads more easily than traditional relational databases. Flexibility: NoSQL databases allow for flexible data modeling, making it easier to handle unstructured or semi-structured data such as social media posts, documents, and sensor data. Performance: NoSQL databases are often faster than traditional relational databases, particularly when handling large volumes of data. Availability: NoSQL databases are designed to be highly available and fault-tolerant, ensuring that data is always accessible, even in the event of hardware or network failures. Cost-effectiveness: NoSQL databases can be more cost-effective than traditional relational databases, particularly for large-scale applications that require significant amounts of data storage and processing. Common Use Cases for NoSQL Databases Web applications: NoSQL databases are often used to power web applications, which require scalability, performance, and flexibility. Big Data: NoSQL databases are commonly used in big data applications, where traditional relational databases can struggle to handle the massive volumes of data involved. Internet of Things (IoT): NoSQL databases are used to store and process data from IoT devices, which can generate massive amounts of data in real time. Real-Time Analytics: NoSQL databases can be used for real-time analytics, enabling businesses to make faster, data-driven decisions. Content Management: NoSQL databases are often used for content management applications, which require the ability to handle unstructured or semi-structured data such as documents, images, and videos. Big Data Technologies Using NoSQL Big data technologies rely on NoSQL databases due to their scalability and ability to handle large volumes of unstructured and semi-structured data. Here are some of the most used big data technologies that leverage NoSQL databases: Hadoop: Hadoop is a popular open-source big data platform that includes the Hadoop Distributed File System (HDFS) for storing and processing large amounts of data, and Apache HBase, a NoSQL column-family database that provides low-latency access to Hadoop data. Cassandra: Apache Cassandra is a highly scalable NoSQL column-family database that is often used in big data applications. Cassandra can handle massive amounts of data across multiple nodes and data centers, making it ideal for distributed systems. MongoDB: MongoDB is a popular document-oriented NoSQL database that is often used in big data applications. MongoDB can store and process large amounts of data, and its flexible data model makes it well-suited for handling unstructured data. Couchbase: Couchbase is a NoSQL document-oriented database that provides a distributed key-value store with high performance and scalability. It is often used in big data applications where real-time data access and processing are critical. Neo4j: Neo4j is a graph database that is often used in big data applications that require the processing of complex relationships between data points. Neo4j is well-suited for applications such as social networks, recommendation engines, and fraud detection systems. Overall, NoSQL databases are a critical component of many big data architectures, enabling organizations to store and process large volumes of data efficiently and effectively. Conclusion NoSQL databases have become increasingly popular in recent years due to their ability to handle large amounts of unstructured or semi-structured data, their scalability, and their high availability. They provide a flexible data model that can adapt to changing data requirements and allow for efficient data processing. NoSQL databases come in various types, including document-oriented, key-value, column-family, and graph databases. Each type has its own strengths and weaknesses, and the choice of the database will depend on the specific requirements of the application. One of the key trade-offs when using NoSQL databases is the CAP theorem, which states that consistency, availability, and partition tolerance cannot be simultaneously guaranteed in a distributed system. NoSQL databases typically prioritize either availability or partition tolerance over consistency, which can lead to data inconsistencies in certain failure scenarios. Overall, NoSQL databases have revolutionized the way we store and process data, particularly in big data applications. They provide a powerful and flexible alternative to traditional relational databases and have become a critical component of many modern data architectures. However, as with any technology, they have their limitations and are not always the best choice for every application. It's important to carefully evaluate the requirements of your application and choose the database that best fits those needs.

In this article, I’ll explain how to use database hooks in your Node.js applications to solve specific problems that might arise in your development journey. Many applications require little more than establishing a connection pool between a server, database, and executing queries. However, depending on your application and database deployments, additional configurations might be necessary. For example, multi-region distributed SQL databases can be deployed with different topologies depending on the application use case. Some topologies require setting properties on the database on a per-session basis. Let’s explore some of the hooks made available by some of the most popular database clients and ORMs in the Node.js ecosystem. Laying the Foundation The Node.js community has many drivers to choose from when working with the most popular relational databases. Here, I’m going to focus on PostgreSQL-compatible database clients, which can be used to connect to YugabyteDB or another PostgreSQL database. Sequelize, Prisma, Knex and node-postgres are popular clients with varying feature sets depending on your needs. I encourage you to read through their documentation to determine which best suits your needs. These clients come with hooks for different use cases. For instance: Connection hooks: Execute a function immediately before or after connecting and disconnecting from your database. Logging hooks: Log messages to stdout at various log levels. Lifecycle hooks: Execute a function immediately before or after making calls to the database. In this article, I’ll cover some of the hooks made available by these clients and how you can benefit from using them in your distributed SQL applications. I’ll also demonstrate how to use hooks to hash a user's password before creation and how to set runtime configuration parameters after connecting to a multi-region database with read replicas. Sequelize The Sequelize ORM has a number of hooks for managing the entire lifecycle of your database transactions. The beforeCreate lifecycle hook can be used to hash a password before creating a new user: JavaScript User.beforeCreate(async (user, options) => { const hashedPassword = await hashPassword(user.password); user.password = hashedPassword; }); Next, I’m using the afterConnect connection hook to set session parameters. With this YugabyteDB deployment, you can execute reads from followers to reduce latencies, and eliminate the need to read from the primary cluster nodes: JavaScript const config = { host: process.env.DB_HOST, port: 5433, dialect: "postgres", dialectOptions: { ssl: { require: true, rejectUnauthorized: true, ca: [CERTIFICATE], }, }, pool: { max: 5, min: 1, acquire: 30000, idle: 10000, }, hooks: { async afterConnect(connection) { if (process.env.DB_DEPLOYMENT_TYPE === "multi_region_with_read_replicas") { await connection.query("set yb_read_from_followers = true; set session characteristics as transaction read only;"); } }, }, }; const connection = new Sequelize( process.env.DATABASE_NAME, process.env.DATABASE_USER, process.env.DATABASE_PASSWORD, config ); By using this hook, each database session in the connection pool will set these parameters upon establishing a new connection: set yb_read_from_followers = true;: This parameter controls whether or not reading from followers is enabled. set session characteristics as transaction read only;: This parameter applies the read-only setting to all statements and transaction blocks that follow. Prisma Despite being the ORM of choice for many in the Node.js community, at the time of writing, Prisma doesn’t contain many of the built-in hooks found in Sequelize. Currently, the library contains hooks to handle the query lifecycle, logging, and disconnecting, but offers no help before or after establishing connections. Here’s how you can use Prisma’s lifecycle middleware to hash a password before creating a user: JavaScript prisma.$use(async (params, next) => { if (params.model == 'User' && params.action == 'create') { params.args.data.password = await hashPassword(params.args.data.password); } return next(params) }) const create = await prisma.user.create({ data: { username: 'bhoyer', password: 'abc123' }, }) To set session parameters to make use of our read replicas, we’ll have to execute a statement before querying our database: JavaScript await prisma.$executeRaw(`set yb_read_from_followers = true; set session characteristics as transaction read only;`); const users = await prisma.user.findMany(); If you need to immediately establish a connection in your connection pool to set a parameter, you can connect explicitly with Prisma to forgo the lazy connection typical of connection pooling. Prisma has the log levels of query , error, info, and warn. Queries can be handled as events using event-based logging: JavaScript const prisma = new PrismaClient({ log: [ { emit: 'event', level: 'query', }, { emit: 'stdout', level: 'error', }, { emit: 'stdout', level: 'info', }, { emit: 'stdout', level: 'warn', }, ], }); prisma.$on('query', (e) => { console.log('Query: ' + e.query); console.log('Params: ' + e.params); console.log('Duration: ' + e.duration + 'ms'); }); This can be helpful in development when working on query tuning in a distributed system. Here’s how you can make use of the beforeExit hook to access the database before exiting: JavaScript const prisma = new PrismaClient(); prisma.$on('beforeExit', async () => { // PrismaClient still available await prisma.issue.create({ data: { message: 'Connection exiting.' }, }) }); Knex Knex is a lightweight query builder, but it does not have the query middleware found in more full-featured ORMs. To hash a password, you can process this manually using a custom function: JavaScript async function handlePassword(password) { const hashedPassword = await hashPassword(password); return hashedPassword; } const password = await handlePassword(params.password); knex('users').insert({...params, password}); The syntax required to achieve a connection hook in the Knex.js query builder is similar to that of Sequelize. Here’s how we can set our session parameters to read from YugabyteDB’s replica nodes: JavaScript const knex = require('knex')({ client: 'pg', connection: {/*...*/}, pool: { afterCreate: function (connection, done) { connection.query('set yb_read_from_followers = true; set session characteristics as transaction read only;', function (err) { if (err) { //Query failed done(err, conn); } else { console.log("Reading from replicas."); done(); } }); } } }); node-postgres The node-postgres library is the most low-level of all of the libraries discussed. Under the hood, the Node.js EventEmitter is used to emit connection events. A connect event is triggered when a new connection is established in the connection pool. Let’s use it to set our session parameters. I’ve also added an error hook to catch and log all error messages: JavaScript const config = { user: process.env.DB_USER, host: process.env.DB_HOST, password: process.env.DB_PASSWORD, port: 5433, database: process.env.DB_NAME, min: 1, max: 10, idleTimeoutMillis: 5000, connectionTimeoutMillis: 5000, ssl: { rejectUnauthorized: true, ca: [CERTIFICATE], servername: process.env.DB_HOST, } }; const pool = new Pool(config); pool.on("connect", (c) => { c.query("set yb_read_from_followers = true; set session characteristics as transaction read only;"); }); pool.on("error", (e) => { console.log("Connection error: ", e); }); There aren’t any lifecycle hooks at our disposal with node-postgres, so hashing our password will have to be done manually, like with Prisma: JavaScript async function handlePassword(password) { const hashedPassword = await hashPassword(password); return hashedPassword; } const password = await handlePassword(params.password); const user = await pool.query('INSERT INTO user(username, password) VALUES ($1, $2) RETURNING *', [params.username, password]); Wrapping Up As you can see, hooks can solve a lot of the problems previously addressed by complicated and error-prone application code. Each application has a different set of requirements and brings new challenges. You might go years before you need to utilize a particular hook in your development process, but now, you’ll be ready when that day comes. Look out for more from me on Node.js and distributed application development. Until then, keep on coding!

In this hands-on lab from ScyllaDB University, you will learn how to use the ScyllaDB CDC source connector to push the row-level changes events in the tables of a ScyllaDB cluster to a Kafka server. What Is ScyllaDB CDC? To recap, Change Data Capture (CDC) is a feature that allows you to not only query the current state of a database’s table but also query the history of all changes made to the table. CDC is production-ready (GA) starting from ScyllaDB Enterprise 2021.1.1 and ScyllaDB Open Source 4.3. In ScyllaDB, CDC is optional and enabled on a per-table basis. The history of changes made to a CDC-enabled table is stored in a separate associated table. You can enable CDC when creating or altering a table using the CDC option, for example: CREATE TABLE ks.t (pk int, ck int, v int, PRIMARY KEY (pk, ck, v)) WITH cdc = {'enabled':true}; ScyllaDB CDC Source Connector ScyllaDB CDC Source Connector is a source connector capturing row-level changes in the tables of a ScyllaDB cluster. It is a Debezium connector, compatible with Kafka Connect (with Kafka 2.6.0+). The connector reads the CDC log for specified tables and produces Kafka messages for each row-level INSERT, UPDATE, or DELETE operation. The connector is fault-tolerant, retrying reading data from Scylla in case of failure. It periodically saves the current position in the ScyllaDB CDC log using Kafka Connect offset tracking. Each generated Kafka message contains information about the source, such as the timestamp and the table name. Note: at the time of writing, there is no support for collection types (LIST, SET, MAP) and UDTs—columns with those types are omitted from generated messages. Stay up to date on this enhancement request and other developments in the GitHub project. Confluent and Kafka Connect Confluent is a full-scale data streaming platform that enables you to easily access, store, and manage data as continuous, real-time streams. It expands the benefits of Apache Kafka with enterprise-grade features. Confluent makes it easy to build modern, event-driven applications, and gain a universal data pipeline, supporting scalability, performance, and reliability. Kafka Connect is a tool for scalably and reliably streaming data between Apache Kafka and other data systems. It makes it simple to define connectors that move large data sets in and out of Kafka. It can ingest entire databases or collect metrics from application servers into Kafka topics, making the data available for stream processing with low latency. Kafka Connect includes two types of connectors: Source connector: Source connectors ingest entire databases and stream table updates to Kafka topics. Source connectors can also collect metrics from application servers and store the data in Kafka topics, making the data available for stream processing with low latency. Sink connector: Sink connectors deliver data from Kafka topics to secondary indexes, such as Elasticsearch, or batch systems, such as Hadoop, for offline analysis. Service Setup With Docker In this lab, you’ll use Docker. Please ensure that your environment meets the following prerequisites: Docker for Linux, Mac, or Windows. Note: running ScyllaDB in Docker is only recommended to evaluate and try ScyllaDB. ScyllaDB open source. For the best performance, a regular install is recommended. 8 GB of RAM or greater for Kafka and ScyllaDB services. docker-compose Git ScyllaDB Install and Init Table First, you’ll launch a three-node ScyllaDB cluster and create a table with CDC enabled. If you haven’t done so yet, download the example from git: git clone https://github.com/scylladb/scylla-code-samples.git cd scylla-code-samples/CDC_Kafka_Lab This is the docker-compose file you’ll use. It starts a three-node ScyllaDB Cluster: version: "3" services: scylla-node1: container_name: scylla-node1 image: scylladb/scylla:5.0.0 ports: - 9042:9042 restart: always command: --seeds=scylla-node1,scylla-node2 --smp 1 --memory 750M --overprovisioned 1 --api-address 0.0.0.0 scylla-node2: container_name: scylla-node2 image: scylladb/scylla:5.0.0 restart: always command: --seeds=scylla-node1,scylla-node2 --smp 1 --memory 750M --overprovisioned 1 --api-address 0.0.0.0 scylla-node3: container_name: scylla-node3 image: scylladb/scylla:5.0.0 restart: always command: --seeds=scylla-node1,scylla-node2 --smp 1 --memory 750M --overprovisioned 1 --api-address 0.0.0.0 Launch the ScyllaDB cluster: docker-compose -f docker-compose-scylladb.yml up -d Wait for a minute or so, and check that the ScyllaDB cluster is up and in normal status: docker exec scylla-node1 nodetool status Next, you’ll use cqlsh to interact with ScyllaDB. Create a keyspace, and a table with CDC enabled, and insert a row into the table: docker exec -ti scylla-node1 cqlsh CREATE KEYSPACE ks WITH replication = {'class': 'NetworkTopologyStrategy', 'replication_factor' : 1}; CREATE TABLE ks.my_table (pk int, ck int, v int, PRIMARY KEY (pk, ck, v)) WITH cdc = {'enabled':true}; INSERT INTO ks.my_table(pk, ck, v) VALUES (1, 1, 20); exit [guy@fedora cdc_test]$ docker-compose -f docker-compose-scylladb.yml up -d Creating scylla-node1 ... done Creating scylla-node2 ... done Creating scylla-node3 ... done [guy@fedora cdc_test]$ docker exec scylla-node1 nodetool status Datacenter: datacenter1 ======================= Status=Up/Down |/ State=Normal/Leaving/Joining/Moving -- Address Load Tokens Owns Host ID Rack UN 172.19.0.3 ? 256 ? 4d4eaad4-62a4-485b-9a05-61432516a737 rack1 UN 172.19.0.2 496 KB 256 ? bec834b5-b0de-4d55-b13d-a8aa6800f0b9 rack1 UN 172.19.0.4 ? 256 ? 2788324e-548a-49e2-8337-976897c61238 rack1 Note: Non-system keyspaces don't have the same replication settings, effective ownership information is meaningless [guy@fedora cdc_test]$ docker exec -ti scylla-node1 cqlsh Connected to at 172.19.0.2:9042. [cqlsh 5.0.1 | Cassandra 3.0.8 | CQL spec 3.3.1 | Native protocol v4] Use HELP for help. cqlsh> CREATE KEYSPACE ks WITH replication = {'class': 'NetworkTopologyStrategy', 'replication_factor' : 1}; cqlsh> CREATE TABLE ks.my_table (pk int, ck int, v int, PRIMARY KEY (pk, ck, v)) WITH cdc = {'enabled':true}; cqlsh> INSERT INTO ks.my_table(pk, ck, v) VALUES (1, 1, 20); cqlsh> exit [guy@fedora cdc_test]$ Confluent Setup and Connector Configuration To launch a Kafka server, you’ll use the Confluent platform, which provides a user-friendly web GUI to track topics and messages. The confluent platform provides a docker-compose.yml file to set up the services. Note: this is not how you would use Apache Kafka in production. The example is useful for training and development purposes only. Get the file: wget -O docker-compose-confluent.yml https://raw.githubusercontent.com/confluentinc/cp-all-in-one/7.3.0-post/cp-all-in-one/docker-compose.yml Next, download the ScyllaDB CDC connector: wget -O scylla-cdc-plugin.jar https://github.com/scylladb/scylla-cdc-source-connector/releases/download/scylla-cdc-source-connector-1.0.1/scylla-cdc-source-connector-1.0.1-jar-with-dependencies.jar Add the ScyllaDB CDC connector to the Confluent connect service plugin directory using a Docker volume by editing docker-compose-confluent.yml to add the two lines as below, replacing the directory with the directory of your scylla-cdc-plugin.jar file. image: cnfldemos/cp-server-connect-datagen:0.5.3-7.1.0 hostname: connect container_name: connect + volumes: + - <directory>/scylla-cdc-plugin.jar:/usr/share/java/kafka/plugins/scylla-cdc-plugin.jar depends_on: - broker - schema-registry Launch the Confluent services: docker-compose -f docker-compose-confluent.yml up -d Wait a minute or so, then access http://localhost:9021 for the Confluent web GUI. Add the ScyllaConnector using the Confluent dashboard: Add the Scylla Connector by clicking the plugin: Fill the “Hosts” with the IP address of one of the Scylla nodes (you can see it in the output of the nodetool status command) and port 9042, which is listened to by the ScyllaDB service. The “Namespace” is the keyspace you created before in ScyllaDB. Notice that it might take a minute or so for the ks.my_table to appear: Test Kafka Messages You can see that MyScyllaCluster.ks.my_table is the topic created by the ScyllaDB CDC connector. Now, check for Kafka messages from the Topics panel: Select the topic, which is the same as the keyspace and table name that you created in ScyllaDB: From the “Overview” tab, you can see the topic info. At the bottom, it shows this topic is on partition 0. A partition is the smallest storage unit that holds a subset of records owned by a topic. Each partition is a single log file where records are written to it in an append-only fashion. The records in the partitions are each assigned a sequential identifier called the offset, which is unique for each record within the partition. The offset is an incremental and immutable number maintained by Kafka. As you already know, the ScyllaDB CDC messages are sent to the ks.my_table topic, and the partition id of the topic is 0. Next, go to the “Messages” tab and enter partition id 0 into the “offset” field: You can see from the output of the Kafka topic messages that the ScyllaDB table INSERT event and the data were transferred to Kafka messages by the Scylla CDC Source Connector. Click on the message to view the full message info: The message contains the ScyllaDB table name and keyspace name with the time, as well as the data status before the action and afterward. Since this is an insert operation, the data before the insert is null. Next, insert another row into the ScyllaDB table: docker exec -ti scylla-node1 cqlsh INSERT INTO ks.my_table(pk, ck, v) VALUES (200, 50, 70); Now, in Kafka, wait for a few seconds and you can see the details of the new Message: Cleanup Once you are done working on this lab, you can stop and remove the Docker containers and images. To view a list of all container IDs: docker container ls -aq Then you can stop and remove the containers you are no longer using: docker stop <ID_or_Name> docker rm <ID_or_Name> Later, if you want to rerun the lab, you can follow the steps and use docker-compose as before. Summary With the CDC source connector, a Kafka plugin compatible with Kafka Connect, you can capture all the ScyllaDB table row-level changes (INSERT, UPDATE, or DELETE) and convert those events to Kafka messages. You can then consume the data from other applications or perform any other operation with Kafka.

What Is Amazon SQS? Amazon SQS (Simple Queue Service) is a message queue service that enables application components to communicate with each other by exchanging messages. This is widely used to build event-driven systems or decouple services on AWS. Features of Amazon SQS Message persistence: Messages are stored in queues until they are delivered or deleted by the sending or receiving endpoints. Guaranteed message delivery: Messages are delivered at least once and in the same order as they are sent. Message redelivery: If a message is not acknowledged, it will be resent up to three times before being deleted from the queue. Visibility timeout: Messages can be set to expire and deleted after a set amount of time, even if they have not been delivered yet. Benefits of Amazon SQS Amazon SQS is a highly reliable and scalable message queuing service that enables you to reliably connect your applications. It provides the following benefits: Low cost of ownership: Amazon SQS is cost-effective due to its pay-per-use pricing model and the ability to use any of the AWS services. High throughput: It can handle more than 1 million messages per second with a latency below 1 ms. Fault tolerance: The service is designed to be highly available and durable with no single point of failure. Security: Amazon SQS uses TLS and message signing when sending messages between clients, as well as authentication mechanisms for clients accessing the service. How To Process Messages in a FIFO Order With SQS Wait, aren’t queues supposed to be First-in-First-out by design? Well, yes, but with SQS, it gets a little complicated. AWS claims they make the best effort to process messages in sequence, but, now and then, there can be instances of messages getting processed out of turn. SQS has a distributed architecture with lots of redundancy. This means there is more than one message store. At runtime, messages are picked up randomly from one of the stores. Let me try and explain this better with an analogy. Suppose a group of three people have gone to purchase tickets at a railway station ticket counter. They decide to stand in three different queues, and whoever gets the tickets first will notify the others so they can come out of the queue—distributed and redundant. Let us assume that all three queues have the same number of people at that instance. As luck would have it, another group of three people come into the station at almost the same time. As they were splitting up into three queues, a person from the second group managed to move past the first group and was ahead in one of the queues. At their turn, the person from the second group got the tickets before the other group—not desired but can happen at times. So, what is the way out? If a strict FIFO order is required, AWS recommends using a FIFO queue. A FIFO SQS, unlike the standard SQS, guarantees strict ordering. As an analogy, suppose you go into a bank and are immediately handed over a token. Token holders are served sequentially, ruling out the possibility of someone getting served out of turn. How Are Messages Processed? Standard SQS queues guarantee “at least once” processing. This means messages won’t be lost and would be processed at least once. But what is “at least once” supposed to mean? Does it mean that messages can be processed more than once? Well, the answer is yes. Let us first look at a message lifecycle. The following are the stages: A message is put onto a queue. It gets picked up by the consumer. Once processed, the consumer deletes it from the queue. Note: At post-processing, the message is not automatically deleted—it has to be explicitly deleted by the consumer. Between stages #2 and #3, the message is “in-flight.” When a message is in-flight, a visibility timeout comes into play that suppresses the message in the queue so it is not processed again. The visibility timeout can be configured, the default being 30 secs. The idea is that a message has to be processed and subsequently deleted from the queue before the visibility timeout expires to avoid duplicate processing. However, there can be times when a message gets stuck while processing, resulting in the visibility timeout expiring and the message getting picked up again by the consumer. Also, it can so happen that, during the delete process, one of the servers gets off the hook, and the message lives on that particular server. When it comes back up, the message gets processed again. So, it is absolutely necessary to design the applications to be idempotent when using a standard SQS. That is, even if a message is processed more than once, it shouldn’t have any business impact. Coming back to our railway station example, let us assume that once a person in a group gets the tickets, he will text all the others in the group. But, while sending the text, if the mobile of one of the receivers gets off the network, that person will not receive the message and will purchase the tickets again. This is a common scenario where a message can be processed more than once. Messages in the FIFO SQS, on the other hand, get processed exactly once. This leads us to another important topic—how are duplicate messages handled? The standard SQS doesn’t care if you put in duplicate messages—the downline application is supposed to be idempotent. The FIFO queue, on the other hand, does not allow duplicates. It creates a deduplication id, which is essentially a hash value based on the payload. However, if the same message has to be processed within a small time window: the default deduplication id will not work, and a custom random deduplication id has to be created and will allow all messages coming in to be processed even if it is exactly the same to one of the previous messages. Which Type of SQS Should You Use? As a rule of thumb, you should always look to use the standard SQS. It is distributed, redundant, and comes with unlimited throughput. After all, it is designed to scale and serve all types of workloads with considerable ease. However, if a strict order is of the utmost importance for the application you are building and you don’t care much about throughput, then, obviously, FIFO will be your best choice.

Event sourcing is not a new term. If you are working in tech, you must have come across event sourcing. Event sourcing is a powerful tool, and it is adapted by many large organizations as their database architectural design. It has the capability to scale up and serve the needs of the modern data industry. In this article, we will understand more about event sourcing and why it’s gaining popularity. We will also discuss the popular question: Is event sourcing going to outgrow databases? What Is Event Sourcing? Event sourcing is an architectural pattern to store data as a sequence of events. An event is nothing but a context to the business operation. For instance, if a customer has requested a refund but you don’t know why, event sourcing gives the context as to why a refund was made. Let’s understand some key terms associated with event sourcing: Events Events represent change in the state of data. These are immutable facts that provide context to the business. Let’s take an example of an e-commerce store. All the data changes will be stored as events, like ProductAdded, ProductOrdered, ProductShipped, PaymentReceived, and so on. Events are recorded in the past tense and provide the source of truth to the current business state. In addition to the context, events also store metadata to provide more information. Event Source Database An event source database, also known as event source db, records all the events in an append-only database. In an event source db, the history of changes is maintained in chronological order. The event source db can also be referred to as event logs. They cannot be changed, as events are immutable. There’s a counter argument, however, that says the event source db can be changed by adding another event — the state of the event source db or its outcome is changing. The event source db for the e-commerce store will record each event along with associated metadata. Suppose there are two events in the Product’s event source db, and a third product is added to the event source db. This added product is not new but is returned by the customer, so the context of the event is returned product and the number of events is updated. The event source db holds all these events in chronological order. The context and chronological order of events provide useful information for in-depth analysis. Streams Event streams provide a complete history of modifications related to a particular event. Events reside in the event source database and event streams represent the order in which events happen. Event streams can be short-lived or long-lived based on the particular scenario. The events within the event stream are identified by a unique numeric value and are incremented as events are updated. You can retrieve the original state of events through these identifiers. In the e-commerce store example, payment is a separate entity/domain object. The payment object has its own unique identifier and its event stream will be: Payment confirmed -> Payment Received -> Refund is requested -> Refund amount deducted. Query View In event sourcing, query models represent logical transitioning of the source write model to read model. They are also referred to as Projections or View Models. In the query view, there are two types of concepts: read model and write model. Let’s recall the example of an e-commerce store where the write model events that are added to the query view are: Order placed, payment received, order dispatched, product deducted, and then we use the query view to generate a summary of all the orders made and the payments received in the read model. Why Do We Need Event Sourcing? Event sourcing is an excellent choice in a variety of applications. Let’s discuss a few scenarios where event sourcing is an acceptable solution. Event sourcing is useful in auditing systems where logs can be stored in chronological order and has on-demand back up option. Traditional methods collect data in specific locations to be used only when needed. By quickly responding to newly available information, an event-driven approach can be more effective. By subscribing to the stream, an organization can receive updates about new occurrences and respond to them immediately. This makes it simpler to model and create intricate business procedures. It is possible to migrate legacy systems to contemporary distributed architectures slowly, eventually replacing particular functionalities with event-sourced services. While writes are directed to the services, the legacy system’s current read pathways can continue to be used. Dependent services can “catch up” when the originating service resumes operation if one goes down. When each service comes back up, synchronization may be accomplished because events are stored in the stream in a specific order. In an event-sourced system, data travels in one direction through separate models to read or update data. Due to the singular responsibility that each part of the data flow has, it makes it easier to reason about data and troubleshoot problems. How Is an Event Source Database (db) Different from Traditional Databases? Data is stored in traditional databases using the CRUD operation, that is, Create, Read, Update, and Delete. Whenever a change happens, the record is updated in the database and it preserves the current state of the system. In all relational and non-relational databases, records can be deleted and the state of the system will be lost. In an event source db, events are immutable; they can’t be deleted or altered. The event source db preserves the history of logs in chronological order. By tracking changes discrepancies between audit data and transactional data are avoided. Just like in CRUD system design, event sourcing stores the events in tables, but in chronological order. Since the data is in order with the latest data at the top, filtering event sourcing is easier as compared to traditional databases. Will Event Sourcing Outgrow Databases? In real-world applications, multiple concurrent users are updating records in the data store, and often data is not updated in all places. This results in inconsistency across data stores. There is no mechanism to store the metadata of history of changes which can be used for in-depth analysis. Event sourcing also provides context to the change happening in the database which helps in answering the business questions. Event sourcing works better with microservices and is reliable to share data between other services. Here’re a few advantages that make event sourcing a better choice than traditional databases: Events can be saved using an append-only operation and are immutable. Tasks that deal with events can run in the background while the user interface, workflow, or process that started them can continue. This, along with the absence of conflict during transaction processing, can greatly enhance the performance and scalability of an application, particularly at the presentation level or user interface. Events are straightforward objects that explain an action that took place together with any additional information needed to fully describe the action that the event represents. A data store is not directly updated by events. They are merely recorded for handling when necessary. This can make administration and implementation simpler. For a domain expert, events often have meaning. However, object-relational impedance mismatch might make it challenging to comprehend complicated database tables. Tables are made up objects that depict the system’s current condition rather than actual events. Because event sourcing does not require updating data store objects directly, it can assist prevent conflicts caused by concurrent modifications. The domain model must still be built to withstand queries that can cause an inconsistent state, though. Tasks respond to events by performing actions as they are raised by the event source db. The jobs and events are separated, which offers flexibility and extensibility. Tasks are aware of the type of event that occurred and its data, but not the operation that caused it. Additionally, each event can be handled by numerous tasks. This makes it possible to integrate additional services and systems that are limited to monitoring new events raised by the event source db with ease. The event sourcing events, however, frequently have a very low level, thus it would be essential to create particular integration events in their place. Challenges With Event Sourcing Despite having a lot of advantages, event sourcing has a lot of challenges as well. Let’s discuss a few challenges associated with event sourcing. The event source db is immutable and serves as a permanent repository for data. Event data should never be modified. Adding a new event to the event source db is the sole option to update an entity to reverse a modification. It can be challenging to mix current events in the store with the new version if the format (rather than the contents) of the persistent events needs to change, possibly during a migration. It could be necessary to introduce new events that utilize the new format or loop through all the existing events making adjustments to bring them into compliance with the new format. To keep both the old and the new event forms, think about using a version stamp on each iteration of the event schema. Querying data or reading data from event source data stores can be difficult, as there’s no standard SQL mechanism. To read the data, the event stream is extracted against the event identifier. The event source db may contain events that were stored by multi-threaded programs and multiple instances of apps. Both the consistency of events in the event source db and the timing of events that have an impact on a particular entity are crucial (the order that changes occur to an entity affects its current state). Every event should have a timestamp to assist avoid problems. Another typical technique is to assign an incremental identification to each event that results from a request. The event source db may reject an event that matches an existing entity identifier and event identifier if two operations attempt to add events for the same entity at the same time. Event sourcing reduces the likelihood of conflicting data changes, but the application must still be able to handle inconsistencies brought on by eventual consistency and the absence of transactions. A customer may need to be informed or a back order may need to be created, for instance, if an event in the data store indicating a reduction in stock inventory occurs while an order is being placed for that item. Once event source systems have been capturing events for some time, another difficulty will arise. It becomes vital to find a technique to handle historical events because, although it’s one thing to record all events that a system has handled, failing to understand that history renders the event log completely useless. The whole event log may need to be reprocessed in order to bring the system’s data universe up to date during system failure recovery events or while migrating derived state stores. Periodic system state snapshots may also be necessary for systems handling large amounts of events, where processing the entire event log again would go beyond any recovery time goals, so that recovery can start from a more recent known good state. Organizations must take into account how events are formed, how that structure can vary over time as the collection of fields changes, and how events with earlier structures might be processed using the current business logic given changes in how the business runs over time. It may be possible to future-proof the event recording by using a defined, extensible event schema, but it may also be necessary to add additional processing rules to the most recent business logic to ensure that earlier event structures can still be understood. Periodic snapshots could also be used to distinguish between significant changes to the event structure, where the expense of maintaining prior events ends up being more than their intrinsic value. Conclusion We have studied the concepts of event sourcing and its advantages and disadvantages in detail. As a final verdict, event sourcing is a great architectural design pattern to store data. However, it can only bring value when used the right way. There are a few scenarios where traditional database technologies are a better option and should be used. Event sourcing is going to be adapted massively in the coming years, but it can’t replace traditional databases. CRUD-based databases are here to stay, and they also serve a huge number of real-world applications. Thanks for reading this article!

SQL Server Management Studio (SSMS) is one of the most proven database administration and management tools. Whether you use SQL Server Management Studio for Database Administrators or just for database development, it's a proven tool. We want to increase our capabilities, adapting this tool to our specific needs. For this, we use a range of SSMS add-ins and extensions. We'll discuss two productivity-focused add-ins that help us achieve more in less time. They're a great way to extend the functionality of the SSMS. They make it more flexible and enhance our SQL codings with autocompletion, snippets, and refactoring. These are SSMS Tools Pack and dbForge SQL Complete. Let's first take a brief overview of both products, then look at features that might interest you. It helps you decide whether SQL Complete is better than SQL Server Management Studio. The Overview of dbForge SQL Complete We start with dbForge SQL Complete, available for SSMS and Microsoft Visual Studio. It offers IntelliSense-like context-based code completion, SQL formatting, and smart refactoring with auto-correction of references. This makes coding up to four times faster. It's suitable for your own use, but it helps to form and unify SQL standards for corporate teamwork. This tool has a powerful T-SQL debugger, tab coloring, and document outline. SQL Complete has many features, and a clean interface — all of that make your work more convenient. Pricing: DBForge SQL Complete is available in three editions: Free Basic (Express), Standard, and Professional. The Express edition is a unique offering that comes completely free of charge. There is no other way to extend the code completion functionality in SSMS for free. SQL Complete can also be purchased in a package called dbForge SQL Tools, which includes fifteen essential products that cover nearly any aspect of SQL Server development, management, and administration. The Overview of SSMS Tools Pack While the second contender, SSMS Tools Pack, doesn't come close to being as versatile and powerful as the first contender, it does offer quite a bit of functionality. It is a SQL Server Management Studio plugin that was created to boost the user's productivity. It's easy to use and delivers a handy SQL editor, CRUD procedure generation, snippets, formatting, convenient search with filtering, and SQL execution history. Like SQL Complete, it also includes features that aren't essential yet, like tab coloring and the ability to export to Excel spreadsheets. Pricing: SSMS Tools Pack is a commercial product with licenses available in Small, Large, and Enterprise team packages. A free version is available for one computer for sixty days. Feature Comparison of dbForge SQL Complete and SSMS Tools Pack To make this comparison, we used the latest versions of tools — SQL Complete v6.12.8 and SSMS Tools Pack v5.5.2. Read the feature descriptions carefully. Some may be far more critical for your particular goals than others. Feature dbForgeSQL Complete SSMSTools Pack Compatibility SSMS integration Yes Yes Visual Studio integration Yes No Improved code quality Find invalid objects Yes No CRUD procedure generation Yes Yes Generation of the CREATE/ALTER script for server objects Yes No Execution Plan Analyzer No Yes Renaming of objects, variables, and aliases Yes No T-SQL Debugger Yes No Run on multiple targets Yes Yes Safe work with document environment and databases Various options for executing statements Yes Yes Execution warnings Yes Yes Execution notifications Yes No Transaction reminder Yes Yes Run At Status Bar Element No Yes Tab coloring Yes Yes Custom SSMS main window title Yes Yes Execution history of SQL statements Yes Yes Tab management Yes Yes Quick Connect Active SQL Editor Window No Yes Document sessions Yes No Operations with data in the SSMS data grid Results Grid data visualizers Yes No Copy Data As from the SSMS grid to XML, CSV, HTML, and JSON Yes No Copy Results Grid headers (column names + types) Yes No Export to Excel from the SSMS Results Grid No Yes Grid aggregates Yes Yes Find in Results Grid Yes Yes Generate Script As from the SSMS data grid Yes Yes Increased coding productivity Context-sensitive suggestion of object names and keywords Yes No Expand SELECT * Yes Yes Object information Yes No Parameter information Yes No SQL snippets Yes Yes New query template No Yes ‘Go to definition’ for database objects Yes Yes Highlighted occurrences of identifiers Yes No Named regions Yes Yes Document Outline window Yes No Unified SQL standards SQL formatting Yes Yes Multiple predefined formatting profiles Yes No Preservation of the original formatting for any selected piece of code Yes No Command-line interface Yes No Settings Import/Export Settings Wizard Yes Yes Quick search for options Yes No Releases Initial release v1.0 (November 19, 2010) v1.0 (May 1, 2008) Latest release (as of September 2021) v6.12 (September 12, 2022) v5.5 (July 1, 2020) Total quantity of releases 133 41 The Verdict of Comparison of dbForge SQL Complete and SSMS Tools Pack We saw that dbForge SQL Complete has more features than SSMS Tools Pack. It offers more to improve productivity, provides a much wider range of data operations, has noticeably more advanced formatting, and supports a command line interface. You should also be aware that SQL Complete is updated more frequently. If you're looking for an SSMS Tools Pack alternative, then SQL Complete is the best solution for you. It's compatible with both Microsoft Visual Studio and SQL Server Management Studio. It doesn't take a lot of effort to see how effective it is.