> For the complete documentation index, see [llms.txt](https://docs.collibra.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.collibra.com/data-quality-self-hosted/apis/using-notebooks-with-data-quality-observability-classic/setting-up-collibra-dq-on-databricks.md).

# Setting up Collibra DQ on Databricks

This topic provides notebook configuration information specific to Databricks. It is meant to supplement the information provided in [Getting started using notebooks with DQ](/data-quality-self-hosted/apis/using-notebooks-with-data-quality-observability-classic/getting-started-using-notebooks-with-dq.md).

## Additional settings for Databricks notebooks

{% stepper %}
{% step %}
Set the following property at either the Spark cluster-level or the SparkSession-level:

```
spark.sql.sources.disabledJdbcConnProviderList='basic,oracle,mssql'
```

{% endstep %}

{% step %}
In the Spark Config section of your Databricks cluster, enter the following configuration property as one key-value pair per line:

```
spark.sql.sources.disabledJdbcConnProviderList
basic,oracle,mssql
```

{% endstep %}

{% step %}
If you have your own Spark data source, set the property in the SparkSession. For example:

```
SparkSession.getActiveSession.get.conf.set("spark.sql.sources.disabledJdbcConnProviderList", "basic,oracle,mssql")
```

{% endstep %}

{% step %}
(Optional) Update the datasource pool size. This step is necessary if you encounter the `PoolExhaustedException` error when calling DQ APIs. Use the following environment variables:

```
SPRING_DATASOURCE_POOL_MAX_WAIT=500
SPRING_DATASOURCE_POOL_MAX_SIZE=30
SPRING_DATASOURCE_POOL_INITIAL_SIZE=5
```

For more information on setting up Databricks environment variables, refer to the official [Databricks documentation](https://docs.databricks.com/aws/en/compute/configure#environment-variables).
{% endstep %}
{% endstepper %}

## Code examples in Databricks

This section provides code samples that you can use in Databricks to perform a variety of tasks in Collibra DQ.

### Import Collibra DQ library

```
import org.apache.spark.sql.SparkSession 
import org.apache.spark.sql.functions._ 
import org.apache.spark.sql.types._ 
import scala.collection.JavaConverters._ 
import java.util.Date
import java.time.LocalDate
import java.text.SimpleDateFormat
import spark.implicits._ 
import java.util.{ArrayList, List, UUID}
// CDQ Imports 
import com.owl.core.Owl 
import com.owl.common.options._
import com.owl.common.domain2._
import com.owl.core.util.OwlUtils
spark.catalog.clearCache
```

### Bring in customer data from another database

**Bring in customer data from a file**

```
val df = (spark.read
.format("csv").option("header", true).option("delimiter", ","
.load(dbfs:/FileStore/nyse.csv")
)
```

**Bring in customer data from a database**

```
val connProps = Map(
"driver" -> "org.postgresql.Driver", 
"user" -> "your-username", 
"password" -> "your-password",
"url" -> "jdbc:postgresql://abc:1234/postgres",
"dbtable" -> "public.example_data") 
//--- Load Spark DataFrame ---//
val df = spark.read.format("jdbc").options(connProps).load display(df)
display(df) // view your data
```

### Specify the location of the DQ metastore

Use the following variables to set up a DQ metastore database location:

```
val pgHost = "xxxx.amazonaws.com" 
val pgDatabase = "postgres" 
val pgSchema = "public"
val pgUser = "???????" 
val pgPass = "????"
val pgPort = "0000"
```

### Create rules and detect breaks

{% hint style="info" %}
If the rules are already created and assigned to a dataset from the UI, calling owlcheck() automatically executes all the rules associated with the given dataset and there is no need to recreate the rule from notebook.
{% endhint %}

```
val dataset = "cdq_notebook_db_rules"
var date = "2018-01-11"
 
// Options
val opt = new OwlOptions()
opt.dataset = dataset
opt.runId = date
opt.host = pgHost
opt.port = pgPort
opt.pgUser = pgUser
opt.pgPassword = pgPass
 
opt.setDatasetSafeOff(false) // to enable historical overwrite of dataset
 
// Create a simple rule and assign it to dataset
val simpleRule = OwlUtils.createRule(opt.dataset)
      simpleRule.setRuleNm("nyse-stocks-symbol")
      simpleRule.setRuleValue("symbol == 'BHK'")
      simpleRule.setRuleType("SQLG")
      simpleRule.setPerc(1.0)
      simpleRule.setPoints(1)
      simpleRule.setIsActive(1)
      simpleRule.setUserNm("admin")
      simpleRule.setPreviewLimit(8)
 
// Create a rule from generic rules that are created from UI:
val genericRule = OwlUtils.createRule(opt.dataset)
    genericRule.setRuleNm("exchangeRule") // this could be any name
    genericRule.setRuleType("CUSTOM")
    genericRule.setPoints(1)
    genericRule.setIsActive(1)
    genericRule.setUserNm("admin")
    genericRule.setRuleRepo("exchangeCheckRule"); // Validate the generic rule name //from UI
    genericRule.setRuleValue("EXCH") // COLUMN associate with the rule
    
        
// Pre Routine 
val cdq = com.owl.core.util.OwlUtils.OwlContext(df, opt)
cdq.removeAllRules(opt.dataset)
.register(opt)
.addRule(simpleRule)
 
// Scan
cdq.owlCheck()
val results = cdq.hoot() // returns object Hoot, not a DataFrame
//See Json Results(Option for downstream processing)
println("--------------Results:----------------\n")
println(results) //optional
 
//Post Routine, See DataFrame Results (Option for downstream processing)
val breaks = cdq.getRuleBreakRows("nyse-stocks-symbol")
println("--------------Breaks:----------------\n")
display(breaks)
 
// Different Options for handling bad records
val badRecords = breaks.drop("_dataset","_run_id", "_rule_name", "owl_id")
display(badRecords)
 
val goodRecords = df.except(badRecords)
display(goodRecords)
```

### Write break records to a Parquet file

```
// Remove the file if it exists
dbutils.fs.rm("/tmp/databricks-df-example.parquet", true) 
breaks.write.parquet("/tmp/databricks-df-example.parquet")
```

The following image shows the code snippet and the result in Databricks:

![Create CDQ Test Rules In DataBricks](https://4121921244-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FWCrP7Albs8DfJjIQxUxw%2Fuploads%2Fgit-blob-a6794605b4d1eefc78df56cfc8a62d7affbc7bf4%2FScreen%20Shot%202022-04-20%20at%202.18.09%20PM.png?alt=media)

![copying rules from one dataset to another](https://4121921244-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FWCrP7Albs8DfJjIQxUxw%2Fuploads%2Fgit-blob-3a41ad6a7aec4c7d95d56a9d3adc5ca96d33449e%2Fdq-reassign-rules-across-datasets.png?alt=media)

Steps to reassign the rules of one dataset to another via the API.

The breaks and the rules can also be viewed on the [Findings page](/data-quality-self-hosted/core-concepts/findings.md) of the Collibra DQ web application.

### Example Profile job

```
val dataset = "cdq_notebook_nyse_profile"
 
val runList = List("2018-01-01", "2018-01-02", "2018-01-03", "2018-01-04", "2018-01-05"
for(runID <- runList {
// Options
val options = new OwlOptions()
options.dataset = dataset
options.host = pgHost
options.port = pgPort
options.pgUser = pgUser
options.pgPassword = pgPass
 
//Scan
val profileOpt = new ProfileOpt
profileOpt.on = true
profileOpt.setShape(true)
profileOpt.setShapeSensitivity(5.0)
profileOpt.setShapeMaxPerCol(10)
profileOpt.setShapeMaxColSize(10)
profileOpt.setShapeGranular(true)
profileOpt.behaviorEmptyCheck = true
profileOpt.behaviorMaxValueCheck = true
profileOpt.behaviorMinValueCheck = true
profileOpt.behaviorNullCheck = true
profileOpt.behaviorRowCheck = true
profileOpt.behaviorMeanValueCheck = true
profileOpt.behaviorUniqueCheck = true
profileOpt.behaviorMinSupport = 5 // default is 4
profileOpt.behaviorLookback = 5
options.profile = profileOpt
 
var date = runId
var df_1 = df.where($"TRADE_DATE"===s"$date")
 
//Scan
val cdq = OwlUtils.OwlContext(df_1, options)
cdq.register(opt)
cdq.owlCheck()
val profile = cdq.profileDF()
profile.show()
}
```

![CDQ Profile Run In Databricks](https://4121921244-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FWCrP7Albs8DfJjIQxUxw%2Fuploads%2Fgit-blob-d5dcc51c29255d671f843538257e035b866f2f2b%2FScreen%20Shot%202022-04-21%20at%2010.30.21%20AM.png?alt=media)

![The Profile result can be viewed in CDQ Web.](https://4121921244-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FWCrP7Albs8DfJjIQxUxw%2Fuploads%2Fgit-blob-226c5077bbebf96428d4aad6195cd5242165b206%2FScreen%20Shot%202022-04-21%20at%2010.27.59%20AM.png?alt=media)

### Example duplicate detection job

```
val dataset = "cdq_notebook_db_dupe"
var date = "2018-01-11"
 
// Options
val options = new OwlOptions()
options.dataset = dataset
options.runId = date
options.host = pgHost
options.port = pgPort
options.pgUser = pgUser
options.pgPassword = pgPass
 
opt.dupe.ignoreCase = true
opt.dupe.on = true
opt.dupe.lowerBound = 99
opt.dupe.include = Array("SYMBOL", "TRADE_DATE")
 
//Scan
val cdq = OwlUtils.OwlContext(df, opt)
cdq.register(options)
cdq.owlCheck()
 
val dupesDf = cdq.getDupeRecords
dupesDf.show()
```

![CDQ Dupes Run In Databricks](https://4121921244-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FWCrP7Albs8DfJjIQxUxw%2Fuploads%2Fgit-blob-124491c82656520cc9e6ce6fdc5d9a6688655080%2FScreen%20Shot%202022-04-21%20at%202.43.27%20PM.png?alt=media)

![Dupes results can be viewed in CDQ Web.](https://4121921244-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FWCrP7Albs8DfJjIQxUxw%2Fuploads%2Fgit-blob-dab09a9d06a1ce5602453bee2ddd8757496fe047%2FScreen%20Shot%202022-04-21%20at%202.39.33%20PM.png?alt=media)

### Example outlier job

```
import scala.collection.JavaConverters._
import java.util
import java.util.{ArrayList, List, UUID}
 
val dataset = "cdq_notebook_db_outlier"
var date = "2018-01-11"
 
// Options
val options = new OwlOptions()
options.dataset = dataset
options.runId = date
options.host = pgHost
options.port = pgPort
options.pgUser = pgUser
options.pgPassword = pgPass
 
opt.dupe.on = false
 
val dlMulti: util.List[OutlierOpt] = new util.ArrayList[OutlierOpt]
val outlierOpt = new OutlierOpt()
outlierOpt.combine = true
outlierOpt.dateColumn = "trade_date"
outlierOpt.lookback = 4
outlierOpt.key = Array("symbol")
outlierOpt.include = Array("high")
outlierOpt.historyLimit = 10
dlMulti.add(outlierOpt)
 
opt.setOutliers(dlMulti)
 
val cdq = OwlUtils.OwlContext(df, opt)
        .register(opt)
 
cdq.owlCheck
val outliers = cdq.getOutliers()
outliers.show
outliers.select("value")
```

### Example ValidateSource job

```
import com.owl.common.options.SourceOpt
import java.util
import java.util.{ArrayList, List}
 
var opt = new OwlOptions()
val dataset = "weather-validateSrc"
opt.setDataset(dataset)
opt.runId = "2018-02-23"
clearPreviousScans(opt)
val src = Seq(
  ("abc", true, 55.5, "2018-02-23 08:30:02"),
  ("def", true, 55.5, "2018-02-23 08:30:02"),
  ("xyz", true, 55.5, "2018-02-23 08:30:02")
).toDF("name", "sunny", "feel-like-temp", "d_date")
 
val target = Seq(
  ("abc", 72, false, 55.5, "2018-02-23 08:30:02"), // true to false
  ("xyz", 72, true, 65.5, "2018-02-23 09:30:02") // 08 to 09
).toDF("name", "temp", "sunny", "feel-like-temp", "d_date")
 
val optSource = new SourceOpt
optSource.on = true
optSource.dataset = dataset
optSource.key = Array("name")
opt.setSource(optSource)
 
//scan
val cdq = OwlUtils.OwlContext(src, target, opt)
  .register(opt)
cdq.owlCheck
 
val breakCountDf = cdq.getSourceBreaksCount()
breakCountDf.show()
```

## Known API Limitations

Collibra DQ activities cannot currently be called independently. DQ Check() function should be called before calling any of the activities. For example, to get the profile DataFrame you should call the following code snippet:

```
cdq.owlCheck()
cdq.getProfileDF()
```