Loading data into Databricks Delta Lake

Create the flow and then configure the Salesforce source to read account data from Salesforce.

1. Click Build > Pipelines to access the Flows view.
2. Click the Add icon.
3. Enter a name for the flow, such as BulkLoadDeltaLake, accept the remaining default selections, and then click Next.
4. Select a registered Data Collector as the authoring Data Collector.
5. Click Save & Open in Canvas.

   An empty flow opens in the flow canvas.

6. From the flow creation help bar, click Select Origin > Salesforce.
7. On the Salesforce tab, enter your Salesforce user name and password.
8. Clear the Subscribe for Notifications checkbox.

   This way, the source runs a query to process existing data and is not subscribed to notifications.

9. Leave the default values for the remaining properties.
10. Click the Query tab and enter the following query for the SOQL Query property so that the source reads only these attributes from the Salesforce account object:

    SELECT Id,
    Name,
    Type,
    BillingStreet,
    BillingCity,
    BillingState,
    BillingPostalCode,
    BillingCountry,
    ShippingStreet,
    ShippingCity,
    ShippingState,
    ShippingPostalCode,
    ShippingCountry,
    Phone,
    Fax
    FROM Account
    WHERE Id > '${OFFSET}'
    ORDER BY Id

11. Leave the default values for the remaining properties.
12. Click the error icon () in the empty flow canvas.

    The properties panel displays the Error Records tab for the flow.

13. Select Discard for the error records.

Next you add and configure the Expression Evaluator processor to clean some of the account data.

1. From the flow creation help bar, click Select Processor > Expression Evaluator.

   The processor is added to the canvas and connected to the source.

2. Select the Expression Evaluator processor in the flow canvas, and then click the Expressions tab.
3. In the Field Expressions section, enter /Type for the Output Field and then enter the following expression for the Field Expression:

   ${str:regExCapture(record:value('/Type'),'(.*) - (.*)',2)}

Add and configure the Databricks target to bulk load the Salesforce data into a Delta Lake table.

1. From the flow creation help bar, click Select Destination > Databricks.

   The target is added to the canvas.

2. Select the target, and then click the Databricks tab.
3. Configure the following properties:

   Property	Value
   JDBC URL	Enter the JDBC URL that the target uses to connect to the Databricks cluster. Remove the PWD parameter from the URL, and then enter the personal access token value for the Token property below. Enter in the following format: jdbc:databricks://<server_hostname>:443/default;transportMode=http :ssl=1;httpPath=sql/protocolv1/o/0/xxxx-xxxxxx-xxxxxxxx;AuthMech=3;
   Token	Enter the personal access token that you generated as a prerequisite in Databricks.
   Table Name	Enter sales.accounts to write the data to the accounts Delta Lake table in the sales database.
   Enable Data Drift	Select to compensate for data drift and automatically create new columns or tables.
   Auto Create Table	Select to automatically create the new accounts table in Delta Lake.

4. Leave the default values for the remaining properties.
5. Click the Staging tab, and then set the Staging Location to Amazon S3.

   The Staging tab defines how the target connects to the specified staging location. This solution uses Amazon S3 as the staging location and assumes that Data Collector runs on an EC2 instance with a configured instance profile. If you prefer, you can configure the target to use an alternate staging location.

6. Configure the following properties:

   Property	Value
   Purge Stage File After Ingesting	Select to enable purging a staged file after its data is successfully written to a Delta Lake table.
   Bucket	Enter the name of the Amazon S3 bucket to write the staged files to.
   Use Instance Profile	Select to use the instance profile assigned to the EC2 instance where Data Collector runs to connect to Amazon S3. If not using instance profiles, clear and enter your AWS secret access key pair.

7. Leave the default values for the remaining properties.

Start a draft run of the flow to move the data from Salesforce to Delta Lake.

1. From the toolbar, click Draft Run > Start Pipeline.
   When the flow successfully starts, you can monitor the health and performance of the flow by viewing real-time statistics and errors as data moves through the flow.

   Because the Salesforce source is configured to read all account data in bulk, the flow automatically stops after reading all account data.

2. Verify that the flow loaded data into the Delta Lake table by running a SQL query in your Databricks notebook.

Create the flow and then configure the MySQL Binary Log source to read CDC information provided by MySQL in binary logs.

1. Click Build > Pipelines to access the Flows view.
2. Click the Add icon.
3. Enter a name for the flow, such as CDCDeltaLake, accept the remaining default selections, and then click Next.
4. Select a registered Data Collector as the authoring Data Collector.
5. Click Save & Open in Canvas.

   An empty flow opens in the flow canvas.

6. From the flow creation help bar, click Select Origin > MySQL Binary Log.
7. On the MySQL Binary Log tab, enter the MySQL server host name and port number.
8. Optionally enter the replication server ID that the source uses to connect to the source MySQL server.

   This solution assumes that the MySQL database is enabled for GTID which does not require that you configure the server ID.

9. Select Start from Beginning.
10. Leave the default values for the remaining properties.
11. Click the Credentials tab and enter the user name and password to connect to MySQL.
12. Click the error icon () in the empty flow canvas.

    The properties panel displays the Error Records tab for the flow.

13. Select Discard for the error records.

Due to the structure of the MySQL binary log records, you need to add several processors to the flow to restructure the record and keep only the necessary fields.

1. From the flow creation help bar, click Select Processor > Stream Selector.

   The processor is added to the canvas.

2. Select the Stream Selector processor in the flow canvas, and then click the Conditions tab.
3. Click the Add icon () to add a condition.
4. Enter the following expression for the condition:

   ${record:value('/Type') == 'DELETE'}

   This condition uses the IBM StreamSets expression language to route records with a DELETE operation in the Type field to the first output stream of the processor. All other records, with an INSERT or UPDATE operation in the Type field, route to the default output stream.

   The configured Conditions tab and the flow should look like this. Note that the Stream Selector processor has two output streams:

   

5. Add a Field Remover processor, and connect the first output stream of the Stream Selector processor to the new processor.
6. Select the Field Remover processor in the flow canvas, and then on the General tab, enter Keep OldData Fields to DELETE for the processor name.
7. Click the Remove/Keep tab.
8. For Action, select Keep Listed Fields, and then enter the following field paths for the Fields property:
   • /OldData
   • /Type

   This configuration keeps only the OldData and Type fields for records with a DELETE operation, and removes all other fields. The flow and the configured Remove/Keep tab should look like this:

   

9. Select the Stream Selector processor in the flow canvas, and then add another Field Remover processor.

   The processor is added to the canvas, connected to the second output stream of the Stream Selector processor.

10. Select the second Field Remover processor in the flow canvas, and then on the General tab, enter Keep Data Fields to INSERT/UPDATE for the processor name.
11. Click the Remove/Keep tab.
12. For Action, select Keep Listed Fields, and then enter the following field paths for the Fields property:
    • /Data
    • /Type

    This configuration keeps only the Data and Type fields for records with an INSERT or UPDATE operation, and removes all other fields. The configured Remove/Keep tab and the flow should look like this:

    

13. Add two Field Flattener processors to the flow, connecting each to one of the Field Remover processors.
14. Select the Field Flattener processor in the stream that keeps the OldData field, and then click the Flatten tab.
15. Configure the following properties with the required values:

    Property	Value
    Flatten	Select Flatten specific fields.
    Fields	Enter /OldData.
    Flatten in Place	Clear the property.
    Target Field	Enter / to write the flattened data to the root field.

16. Leave the default values for the remaining properties.
17. Select the second Field Flattener processor in the stream that keeps the Data field, and then configure it the same way as the first Field Flattener processor, except enter /Data for the Fields property.

Add and configure the Databricks target to merge the changed data to a Delta Lake table.

1. From the flow creation help bar, click Select Destination > Databricks.

   The target is added to the canvas.

2. Use your cursor to connect both Field Flattener processors to the target.
3. Select the target, and then click the Databricks tab.
4. Configure the following properties:

   Property	Value
   JDBC URL	Enter the JDBC URL that the target uses to connect to the Databricks cluster. Remove the PWD parameter from the URL, and then enter the personal access token value for the Token property below. Enter in the following format: jdbc:databricks://<server_hostname>:443/default;transportMode=http :ssl=1;httpPath=sql/protocolv1/o/0/xxxx-xxxxxx-xxxxxxxx;AuthMech=3;
   Token	Enter the personal access token that you generated as a prerequisite in Databricks.
   Table Name	Enter customers_cdc to write the changed data to a customers_cdc table in the default delta database.
   Enable Data Drift	Select to compensate for data drift and automatically create new columns or tables.
   Auto Create Table	Select so that the target can automatically create the new customers_cdc table in Delta Lake.

5. Leave the default values for the remaining properties.
6. Click the Staging tab, and then set the Staging Location to Amazon S3.

   The Staging tab defines how the target connects to the specified staging location. This solution uses Amazon S3 as the staging location and assumes that Data Collector runs on an EC2 instance with a configured instance profile. If you prefer, you can configure the target to use an alternate staging location.

7. Configure the following properties:

   Property	Value
   Purge Stage File After Ingesting	Select to enable purging a staged file after its data is successfully written to a Delta Lake table.
   Bucket	Enter the name of the Amazon S3 bucket to write the staged files to.
   Use Instance Profile	Select to use the instance profile assigned to the EC2 instance where Data Collector runs to connect to Amazon S3. If not using instance profiles, clear and enter your AWS secret access key pair.

8. Leave the default values for the remaining properties.
9. Click the Data tab.
10. Select Merge CDC Data.

    Enabling this property configures the target to use the MERGE command to insert, update, or delete the changed data in Delta Lake tables as appropriate.

11. Configure the following properties for the Key Columns section.

    The target uses the key columns to evaluate the MERGE condition.

    Property	Value
    Table	Enter customers_cdc.
    Key Columns	Enter customer_id.

Start a draft run of the flow to move the changed data from MySQL binary logs to Delta Lake.

1. From the toolbar, click Draft Run > Start Pipeline.
   When the flow successfully starts, you can monitor the health and performance of the flow by viewing real-time statistics and errors as data moves through the flow.
2. Next, verify that the flow loaded the data into the target table in Delta Lake by running a SQL query in your Databricks notebook.
3. Verify that the flow successfully applies update operations to the Delta Lake table by running the following command on the MySQL database to update one of the rows:

   update retail.customers_cdc set address='10 Downing ST' where customer_id=6;

   Then in your Databricks notebook, verify that the Delta Lake table has been updated with the changed address for that customer ID.

4. Verify that the flow successfully applies delete operations to the Delta Lake table by running the following command on the MySQL database to delete one of the rows:

   delete from retail.customers_cdc where customer_id=7;

   Then in your Databricks notebook, verify that the row for that customer ID has been deleted from the Delta Lake table.

5. Click the Stop icon () to stop the flow.