<jupyter_start><jupyter_text>Some Helper Functions / Vars<jupyter_code>%scala
val workspaceID = if (dbutils.notebook.getContext.tags("orgId") == "0") {
  dbutils.notebook.getContext.tags("browserHostName").split("\\.")(0)
} else dbutils.notebook.getContext.tags("orgId")

def pipReport(db: String): DataFrame = {
  val basePip = spark.table(s"${db}.pipeline_report")
    .filter('organization_id === workspaceID)
    .orderBy('Pipeline_SnapTS.desc, 'moduleID)
    .withColumn("fromTSt", from_unixtime('fromTS.cast("double") / lit(1000)).cast("timestamp"))
    .withColumn("untilTSt", from_unixtime('untilTS.cast("double") / lit(1000)).cast("timestamp"))
    .drop("runStartTS", "runEndTS", "dataFrequency", "lastOptimizedTS", "vacuumRetentionHours", "inputConfig", "parsedConfig")
  
  val pipReportColOrder = "organization_id, moduleID, moduleName, primordialDateString, fromTSt, untilTSt, status, recordsAppended, fromTS, untilTS, Pipeline_SnapTS, Overwatch_RunID".split(", ")
  TransformFunctions.moveColumnsToFront(basePip, pipReportColOrder)
}<jupyter_output><empty_output><jupyter_text>Setup Widgets For Simple Adjustments Or Job Configs* Initiallize the widgets if running interactively* Pull the widgets into usable variables to construct the config<jupyter_code>%scala
// dbutils.widgets.removeAll
// dbutils.widgets.text("storagePrefix", "", "1. ETL Storage Prefix")
// dbutils.widgets.text("etlDBName", "overwatch_etl", "2. ETL Database Name")
// dbutils.widgets.text("consumerDBName", "overwatch", "3. Consumer DB Name")
// dbutils.widgets.text("secretsScope", "my_secret_scope", "4. Secret Scope")
// dbutils.widgets.text("dbPATKey", "my_key_with_api", "5. Secret Key (DBPAT)")
// dbutils.widgets.text("ehKey", "overwatch_eventhub_conn_string", "6. Secret Key (EH)")
// dbutils.widgets.text("ehName", "my_eh_name", "7. EH Topic Name")
// dbutils.widgets.text("primordialDateString", "2021-04-01", "8. Primordial Date")
// dbutils.widgets.text("maxDaysToLoad", "60", "9. Max Days")
// dbutils.widgets.text("scopes", "all", "A1. Scopes")
%scala
val storagePrefix = dbutils.widgets.get("storagePrefix")
val etlDB = dbutils.widgets.get("etlDBName")
val consumerDB = dbutils.widgets.get("consumerDBName")
val secretsScope = dbutils.widgets.get("secretsScope")
val dbPATKey = dbutils.widgets.get("dbPATKey")
val ehName = dbutils.widgets.get("ehName")
val ehKey = dbutils.widgets.get("ehKey")
val primordialDateString = dbutils.widgets.get("primordialDateString")
val maxDaysToLoad = dbutils.widgets.get("maxDaysToLoad").toInt
val scopes = if (dbutils.widgets.get("scopes") == "all") {
  "audit,sparkEvents,jobs,clusters,clusterEvents,notebooks,pools".split(",")
} else dbutils.widgets.get("scopes").split(",")

if (storagePrefix.isEmpty || consumerDB.isEmpty || etlDB.isEmpty || ehName.isEmpty || secretsScope.isEmpty || ehKey.isEmpty || dbPATKey.isEmpty) {
  throw new IllegalArgumentException("Please specify all required parameters!")
}
%sql
drop database overwatch cascade;
drop database overwatch_etl cascade;
%scala
dbutils.fs.rm("/tmp/overwatch01/8322965595968751/ehState/rawEventsCheckpoint",recurse=true)
%scala
// If first run this should be empty
display(dbutils.fs.ls(s"${storagePrefix}/${workspaceID}").toDF)<jupyter_output><empty_output><jupyter_text>Construct the Overwatch Params and Instantiate Workspace<jupyter_code>%scala
private val dataTarget = DataTarget(
  Some(etlDB), Some(s"${storagePrefix}/${workspaceID}/${etlDB}.db"), Some(s"${storagePrefix}/global_share"),
  Some(consumerDB), Some(s"${storagePrefix}/${workspaceID}/${consumerDB}.db")
)

private val tokenSecret = TokenSecret(secretsScope, dbPATKey)
private val ehConnString = dbutils.secrets.get(secretsScope, ehKey)

private val ehStatePath = s"${storagePrefix}/${workspaceID}/ehState"
private val badRecordsPath = s"${storagePrefix}/${workspaceID}/sparkEventsBadrecords"
private val azureLogConfig = AzureAuditLogEventhubConfig(connectionString = ehConnString, eventHubName = ehName, auditRawEventsPrefix = ehStatePath)
private val interactiveDBUPrice = 0.56
private val automatedDBUPrice = 0.26

val params = OverwatchParams(
  auditLogConfig = AuditLogConfig(azureAuditLogEventhubConfig = Some(azureLogConfig)),
  dataTarget = Some(dataTarget), 
  tokenSecret = Some(tokenSecret),
  badRecordsPath = Some(badRecordsPath),
  overwatchScope = Some(scopes),
  maxDaysToLoad = maxDaysToLoad,
  databricksContractPrices = DatabricksContractPrices(interactiveDBUPrice, automatedDBUPrice),
  primordialDateString = Some(primordialDateString)
)

private val args = JsonUtils.objToJson(params).compactString
val workspace = if (args.length != 0) {
  Initializer(Array(args), debugFlag = true)
} else { 
  Initializer(Array()) 
}<jupyter_output><empty_output><jupyter_text>Show the Config StringsThis is a good time to run the following commands for when you're ready to convert this to run as a job as a main class<jupyter_code>%scala
JsonUtils.objToJson(params).escapedString
%scala
JsonUtils.objToJson(params).compactString<jupyter_output><empty_output><jupyter_text>Execute The Pipeline<jupyter_code>%scala
Bronze(workspace).run()
%scala
Silver(workspace).run()
%scala
Gold(workspace).run()<jupyter_output><empty_output><jupyter_text>Show The Run Report<jupyter_code>%scala
display(
  pipReport(etlDB)
)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1387218045112782, 'fraction_numerical': 0.010338345864661654, 'mean_word_length': 7.379527559055118, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '229599', 'license': 'MIT', 'n_tokens_mistral': 1848, 'n_tokens_neox': 1759, 'n_words': 332, 'path': 'databricks/notebooks/azure_runner_docs_example.ipynb', 'repo_name': 'lordlinus/one-click-databricks', 'size': '5320'}