--- title: "Streaming" canonical: "https://docs.vaultspeed.com/space/VPD/3080028173/Streaming" format: markdown --- VaultSpeed supports real-time data loading based on Kafka streams with the help of Spark Structured Streaming and the Databricks platform. > Macro (toc) ## Big picture The flow for the real-time data loading is as follows: Kafka topics read produced data. Code running in Databricks subscribes to those Kafka topics and consumes the messages produced by it. Data coming in is processed and pushed to the target database using the Hikari connection pool and a custom VaultSpeed JDBC sink. This means that the data can come from any type of system, as long as it produces Kafka streams. And the target database can be any of the supported database platforms by VaultSpeed. The Databricks cluster is only used as the runtime engine to process the data. An extra feature build into this, is that in the VaultSpeed extract layer logic, data gets delayed (Watermarked), so later on, the VaultSpeed Staging layer logic can use that data for referential integrity resolution. Incidents are handled by Apache spark by configuring checkpoint locations, which use the file system. ![image](media://59529246-dbe2-4773-89cd-207070c0535a) With the streaming solution, VaultSpeed will generate multiple zip files with your code. One with the DDL code to deploy on the target database and one with the Scala files to deploy on the Databricks cluster. Each Scala file represents a single table and will create a continuously running job on Databricks. ## Setup ### Source Setup in VaultSpeed In VaultSpeed, on the parameters of your streaming source, in the “Important parameters” section, you need to set STREAMING_SOURCE = Y. This will eventually generate Databricks Scala code for this source. There are two additional parameters in the same section that are related to Spark Structured Streaming: - WATER_MARK_SIZE - Time in seconds, which defines how long records in the Extraction layer should be delayed, so they can be reused in fixing referential integrity mismatches. (Default value 120 seconds). Since real-time loading will load each object individually, this makes sure that records with foreign-key to primary-key relations will be correctly resolved in the data vault link objects. - ERROR_RECORD_RETENTION_TIME - Time in seconds, which defines how long records in error tables should be kept before they are removed. (Default value 120 seconds) The third parameter related to this can be found in the “Schema parameters” section: - SCHEMA_CDC_KAFKA - String that defines what kind of topic prefix is used in Kafka. (Default value PROJECT_CDC_KAFKA, nullable) With these parameters setup, you can model your source and data vault models just as you would for your non-streaming sources in VaultSpeed. ### Code Generation You can generate code for your streaming source in VaultSpeed in the normal way you would for your other sources. Only this time the ETL-generation part will generate two zip files with code. One will get its target technology automatically set to Databricks SQL, which should be deployed to your Databricks cluster. The other one will have the default target technology of your chosen target database platform and can be deployed as usual to your target database. In the case where your default target technology already is Apache Spark, only one zip file is created. ### Code Deployment To use the Automatic Deployment module of VaultSpeed for your Databricks code, a database link of the type “Apache Spark” should be configured. In your “connections.properties” file of your agent, you should use the following settings to configure a database link to a Databricks cluster: | | | | --- | --- | | .url | Connection to cluster Url | | .token | The access token | | .host | Host of Databricks instance. | | .path | Path to the folder where code should be deployed. | Here is an example of a Databricks connection in a “connections.properties” file. ``` # Databricks config databricks.url="jdbc:spark://.azuredatabricks.net:443/default;transportMode=http;ssl=1;httpPath=sql/protocolv1/o/2546447147849772/1001-055839-parry707;AuthMech=3;UID=token;PWD=" databricks.token= databricks.host=https://.azuredatabricks.net databricks.path=/streaming/ ``` ### Kafka A Kafka connector needs to be configured for Databricks to use it. We do not take responsibility for its setup, but there are a couple of things that we have noticed during our development and testing. `Max Rows Per Batch` - We assume that in a production environment, it should be set to `1`. But since in the description, it's said that it's used for pooling new data, not for producing it, there is a chance that our assumption is incorrect, and it can be left as default value. `Topic Prefix` - Prefix that will be added to on top of the table names to create Kafka topics names. It should be the same as the `SCHEMA_CDC_KAFKA` parameter in the VaultSpeed application. ### Databricks You need to have the Databricks cluster configured beforehand to be able to run the generated Scala code. #### Libraries required There are some libraries required that you need to import into your Databricks cluster: - HikariCP_4_0_3.jar - can be found on the mvn repository website: [HikariCp.](https://mvnrepository.com/artifact/com.zaxxer/HikariCP/4.0.3) - spark_structured_streaming_jdbc_sink.jar - this can be downloaded from VaultSpeed on the DDL/ETL generation page when you are generating code for your streaming source ![image](media://43a65266-2cdb-470f-bf74-b1e16ddfa5a5)   #### Environment variables These are some standard environment variables you can use for your Databricks setup ``` TARGET_CONN_PASSWORD= TARGET_CONN_MAXIMUM_POOL_SIZE=36 TARGET_CONN_AUTO_COMMIT=true TARGET_CONN_CONNECTION_TIMEOUT=10000 SOURCE_KAFKA_BOTSTRAP_SERVERS= TARGET_CONN_TRANSACTION_ISOLATION=TRANSACTION_READ_UNCOMMITTED TRIGGER_PROCESSING_MILS=1024 TARGET_CONN_MAX_LIFETIME=60000 TARGET_CONN_MINIMUM_IDLE=8 TARGET_JDBC_URL=jdbc:postgresql:/// CHECK_POINT_LOCATION= TARGET_CONN_USERNAME=postgres TARGET_CONN_IDLE_TIMEOUT=30000 TARGET_JDBC_DRIVER=org.postgresql.Driver SOURCE_KAFKA_SCHEMA_REGISTRY=http:// ``` #### Optional scheduler An optional job scheduler can assign different weights to each type of processing. This is defined in Spark your config: ``` spark.scheduler.allocation.file /dbfs/configs/vaulspeedscheduler.xml ``` Example of this config: ``` // dbutils.fs.help() // dbutils.fs.mkdirs("configs") // dbutils.fs.ls("/configs") dbutils.fs.put("/configs/vaulspeedscheduler.xml", """ FIFO 3 0 FIFO 3 0 FIFO 3 0 FIFO 2 0 FIFO 2 0 FIFO 2 0 FIFO 4 0 FIFO 1 0 FIFO 1 0 FIFO 1 0 """, true) ``` #### Machine size The Machine size is going to depend on how many jobs are handled. Generally, running streaming jobs is resource hungry and should be scaled up until running notebooks are no longer crashing with `Concurrency` or `Out of Memory` exceptions. The cluster we used for a relatively small model was: - 4 Workers - 112 GB Memory - 32 Cores - 6 DBU - 1 Driver - 28 GB Memory - 8 Cores - 1,5 DBU - No Autoscaling which might look quite a lot, but Spark Structured Streaming is just that demanding. #### Things to note If in case you have to rerun streaming notebooks and they raise the issue that `Streams have been closed before others could complete`, there is a chance that data in checkpoint locations causes some conflict. You can clean checkpoint locations by calling the below command in the cluster. ``` dbutils.fs.rm("path/to/checkpoint/location", true) ``` #### Jobs VaultSpeed generates SQL and Scala files for Databricks, on which you should create jobs. These jobs need to have [Recovery from query failures](https://docs.databricks.com/spark/latest/structured-streaming/production.html#recover-from-query-failures). This is a customer responsibility and VaultSpeed does not generate code to create those jobs. ## Known Limitations There are a few limitations to the current implementation. - Technologies - Currently, Snowflake is not supported as a target (due to an incompatibility in the calls between Databricks and the Snowflake API to run the code on Snowflake). - Object Types - Not all objects types are supported yet. The data vault objects types that are supported are: - Hub and satellite table - Link table - “Many to Many” link table - Reference table - Satellite table - Non-Historized link table - Data Types - Numeric Datatypes without a precision can’t be handled by Kafka. The workaround is to add a precision to them. - Customer defined datatypes aren’t supported by the Kafka JDBC Driver so those are not supported in the solution - Bytes are incorrectly read by Kafka - Parameters - REMOTE_JOURNALING_TABLES are not available for Streaming - Initial load - The initial load for your streaming source should not be run using the streaming/Databricks platform, but just as you would for your other non-streaming sources - An important thing to note here is that this initial load should be empty (meaning no actual data will be loaded). It is only used to populate the Null and Unknown records.