> 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/run-dq-jobs-programmatically.md). # Run DQ jobs programmatically This topic contains examples of common tasks using notebooks. If you are new to using notebooks with Collibra DQ, go to [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). ## Load a table The following example loads a table using SparkJDBC. ``` //--- Configure Table From Database ---// val connProps = Map ( "driver" -> "org.postgresql.Driver", "user" -> s"${user}", "password" -> s"${pass}", "url" -> s"jdbc:postgresql://${host}:${port}/${database}", "dbtable" -> "cdq_test.nyse" ) //--- Load Spark DataFrame ---// val jdbcDF = spark.read.format("jdbc").options(connProps).load jdbcDF.show ``` ## Configure run options The following example connects to the DQ metastore and sets run options. ``` import com.owl.common.options._ import com.owl.core.util.OwlUtils val opt = new OwlOptions //--- Collibra DQ Metastore ---// opt.host = s"$cdqHost" opt.port = s"xxxx/postgres?currentSchema=public" opt.pgUser = s"$cdqUser" opt.pgPassword = s"$cdqPass" //--- Run Options ---// opt.dataset = "cdq_test.nyse" opt.runId = "2018-01-10" opt.datasetSafeOff = true val cdq = OwlUtils.OwlContext(jdbcDF, opt) ``` ## Register and run profile The following example registers with the DQ catalog and runs the profile dataset. ``` //--- Register with Collibra DQ Catalog ---// cdq.register(opt) cdq.owlCheck() //--- Profile Dataset ---// val profile = cdq.profileDF profile.show ``` Notice that DQ returns results as Dataframes. This abstraction allows you to ignore all domain objects and custom types and interact with a scaleable generic result set using common protocols like "where" or "filter" or "save" or "write" all with parallel operations. ``` +--------------+-----+-------+-----------+---+---+--------+-----------+------+----+------+-------+-------+------+----+---------+ | column|nulls|empties|cardinality|min|max|is_mixed|mixed_ratio| Int|Long|String|Decimal|Boolean|Double|Date|Timestamp| +--------------+-----+-------+-----------+---+---+--------+-----------+------+----+------+-------+-------+------+----+---------+ | tenant_id| 0| 0| 60| 0| 9| false| 0.0|100000| 0| 0| 0| 0| 0| 0| 0| | a11| 0| 0| 1|a11|a11| false| 0.0| 0| 0|100000| 0| 0| 0| 0| 0| | a10| 0| 0| 1|a10|a10| false| 0.0| 0| 0|100000| 0| 0| 0| 0| 0| | account_type| 0| 0| 3| 02| 06| false| 0.0|100000| 0| 0| 0| 0| 0| 0| 0| | a13| 0| 0| 1|a13|a13| false| 0.0| 0| 0|100000| 0| 0| 0| 0| 0| |security_alias| 0| 0| 3| 0| 2| false| 0.0|100000| 0| 0| 0| 0| 0| 0| 0| | a12| 0| 0| 1|a12|a12| false| 0.0| 0| 0|100000| 0| 0| 0| 0| 0| +--------------+-----+-------+-----------+---+---+--------+-----------+------+----+------+-------+-------+------+----+---------+ ``` {% hint style="info" %} While the spark `DF.show()` command provides a convenient output format, you may prefer the additional features provided by the DQ UI, such as trend analysis, data drift, data relationships and more. {% endhint %} ## Detect duplicates This following example detects repeated or duplicate records in a table. Be sure to set the lowerBound to the percent fuzzy match you are willing to accept, commonly 87% or higher. ``` opt.dupe.on = true opt.dupe.lowerBound = 99 opt.dupe.include = Array("SYMBOL", "EXCH") val df1Day = jdbcDF.where("TRADE_DATE = '2018-01-10' ") val cdq = OwlUtils.OwlContext(df1Day, opt) cdq.owlCheck() val dupesPreview = cdq.getDupesPreview() val records = cdq.getRecordsDF dupes.show // rdd collect dupes.rdd.collect.foreach(println) // records linked together for remediation cdq.getDupeRecords.show ``` ## Detect outliers The following outlier detection example compares the current day to a baseline of days in the historical dataframe. ``` val outlierOpt = new OutlierOpt() outlierOpt.combine = true outlierOpt.dateColumn = "trade_date" outlierOpt.lookback = 4 outlierOpt.key = Array("symbol") outlierOpt.timeBin = OutlierOpt.TimeBin.DAY outlierOpt.historyLimit = 10 dlMulti.add(outlierOpt) opt.setOutliers(dlMulti) val cdq = OwlUtils.OwlContext(df, opt) .register (opt) cdq.setHistoricalDF(dfHistory) cdq.owlCheck() val outliers = cdq.getOutliers() outliers.show ``` ### Ingest files The following example ingests a series of files. ``` /////////////////////////////////////////////////////////////////////////// // USE CASE - Ingesting Intraday Files // /////////////////////////////////////////////////////////////////////////// // Part of your pipeline includes the ingestion of files that have the date // and hour encoded in the file name. How do you process those files using Collibra DQ? // // Format: ___.csv // // Build up a data structure containg the files you want to process (here we // just use a simple list, but you may want to be pulling from a pubsub // queue, AWS bucket, etc...). Here we just use a simple file list of 6 // hours of trade position data. val position_files = List( new File(getClass.getResource("/position_file_2019_11_03_08.csv").getPath), new File(getClass.getResource("/position_file_2019_11_03_09.csv").getPath), new File(getClass.getResource("/position_file_2019_11_03_08.csv").getPath), new File(getClass.getResource("/position_file_2019_11_03_09.csv").getPath), new File(getClass.getResource("/position_file_2019_11_03_10.csv").getPath), new File(getClass.getResource("/position_file_2019_11_03_11.csv").getPath), new File(getClass.getResource("/position_file_2019_11_03_12.csv").getPath), new File(getClass.getResource("/position_file_2019_11_03_13.csv").getPath), new File(getClass.getResource("/position_file_2019_11_03_14.csv").getPath)) // Create your spark session. val spark = SparkSession.builder .master("local") .appName("test") .getOrCreate() // Configure Collibra DQ val opt = new OwlOptions opt.dataset = "positions" opt.load.delimiter = "," opt.spark.master = "local[1]" opt.dupe.on = true opt.dupe.include = Array("ticker", "cid") opt.outlier.on = true opt.outlier.key = Array("cid") opt.outlier.timeBin = TimeBin.HOUR // Customize this to only process a subset of the data. opt.load.fileQuery = "select * from dataset" position_files.foreach { file: File => // Tell Collibra DQ where to find the file. opt.load.filePath = file.getPath // Parse the filename to construct the run date (-rd) that will be passed // to Collibra DQ. val name = file.getName.split('.').head val parts = name.split("_") val date = parts.slice(2, 5).mkString("-") val hour = parts.takeRight(1).head // Must be in format 'yyyy-MM-dd' or 'yyyy-MM-dd HH:mm'. val rd = s"${date} ${hour}" // Tell Collibra DQ to process data opt.runId = rd // Create a DataFrame from the file. val df = OwlUtils.load(opt.load.filePath, opt.load.delimiter, spark) // Instantiate an OwlContext with the dataframe and our custom configuration. val cdq = OwlUtils.OwlContext(df, spark, opt) // Make sure Collibra DQ has catalogued the dataset. cdq.register(opt) // Let Collibra DQ do the rest! cdq.owlCheck() } ``` For information on rules, go to [Rules in notebooks](/data-quality-self-hosted/apis/using-notebooks-with-data-quality-observability-classic/rules-in-notebooks.md).