Cơ chế xử lý linh động trên Azure Databricks

1.  val storageAccountName = "<your_storage_account_name>"  
2.  val storageAccountKey = "<your_storage_account_key>"  
3.  // Get information from parameters
4.  val containerName = dbutils.widgets.get("containerName")  
5.  val fileName = dbutils.widgets.get("fileName")  
6.  val outputFile = dbutils.widgets.get("outputFile")  
7.  val fileType = dbutils.widgets.get("fileType")  
8.    
9.    
10. // Configure connection string
11. // In reality, you should use Key Vault instead of using key directly
12. spark.conf.set("fs.azure.account.key." + storageAccountName + ".blob.core.windows.net", storageAccountKey)  
13.   
14. // Remove the file if it exists  
15. dbutils.fs.rm(outputFile, true)  
16. 
17. // Download csv file
18. if (fileType == "csv") {  
19.     val dataDf = spark.read  
20.     .option("header","true")  
21.     .option("inferSchema", "true")  
22.     .csv("wasbs://" + containerName + "@"+ storageAccountName+ ".blob.core.windows.net/" + fileName)        
23.     dataDf.write.parquet(outputFile)  
24. } else if (fileType == "json") {  
25. // Download json file
26.     val dataDf = spark.read.json("wasbs://" + containerName + "@"+ storageAccountName+ ".blob.core.windows.net/" + fileName)  
27.     dataDf.write.json(outputFile)  
28. } else {  
29. // Download text file
30.     val dataDf = spark.read.textFile("wasbs://" + containerName + "@"+ storageAccountName+ ".blob.core.windows.net/" + fileName)  
31.     
32.     val result = dataDf.collect().mkString("\n")  
33.     dbutils.notebook.exit(result)  
34. }

1.  val storageAccountName = "<your_storage_account_name>"  
2.  val storageAccountKey = "<your_storage_account_key>"  
3.  // Get information from parameters
4.  dataPath = dbutils.widgets.get("dataPath")  
5.  output_container_name = dbutils.widgets.get("output_container_name")  
6.  outputPath = dbutils.widgets.get("outputPath")  
7.    
8.  # Configure blob storage account access key globally  
9.  # In reality, you should use Key Vault instead of using key directly
10. spark.conf.set("fs.azure.account.key.%s.blob.core.windows.net" % storage_name, storage_key)  
11.   
12. df = spark.read.parquet(dataPath)  
13.   
14. output_container_path = "wasbs://%s@%s.blob.core.windows.net/" % (output_container_name, storage_name)  
15. output_blob_folder = "%stmpFolder" % output_container_path  
16.   
17. # write the dataframe as a single file to blob storage  
18. (df  
19.  .coalesce(1)  
20.  .write  
21.  .mode("overwrite")  
22.  .option("header", "true")  
23.  .format("com.databricks.spark.csv")  
24.  .save(output_blob_folder))  
25.   
26. files = dbutils.fs.ls(output_blob_folder)  
27. output_file = [x for x in files if x.name.startswith("part-")]  
28.   
29. dbutils.fs.mv(output_file[0].path, "%s" % (output_container_path + outputPath))  

1.  import scala.util.matching.Regex  
2.  import org.apache.spark.sql.functions._  
3.    
4.  // define function check  
5.  def isPhoneNumberFormat() = udf((data: String) => {  
6.    var pattern: Regex = "^[+]*[(]84[)]\\d{8,10}$".r  
7.    if (pattern.findFirstMatchIn(data).mkString.length() > 0) {  
8.      1 // correct pattern  
9.    } else {  
10.     val data2 = data.replaceAll("[+()-]|\\s", "") // remove  special character  
11.     if (data2.length() > 11 | data2.length() < 10) {  
12.       0 // not a format of phone number  
13.     } else {  
14.       100 // can be corrected  
15.     }  
16.   }  
17. })  
18. // define corrected function  
19. def correctPhoneNumber() = udf((data: String, check: Int) => {  
20.   if (check == 1 | check == 0) {  
21.     data.toString // no need to correct  
22.   } else {  
23.     if (check == 100) {  
24.       // replace first digit to +(84)  
25.       data.replaceAll("[+()-]|\\s", "").replaceFirst("[0-9]", "+(84)").toString  
26.     } else {  
27.       data  
28.     }  
29.   }  
30. })  
31. // read paramter  
32. val dataPath = dbutils.widgets.get("dataPath")  
33. val colName = dbutils.widgets.get("phoneCol")  
34. val keyCol = dbutils.widgets.get("keyCol")  
35. val tmpColumn = dbutils.widgets.get("tmpColumn")  
36.   
37. var df = spark.read.parquet(dataPath)  
38. 
39. // execute check, return data frame with 3 column: key Column, corrected column, check column  
40. df = df.withColumn(tmpColumn, isPhoneNumberFormat()(df(colName)))  
41. df = df.withColumn(colName, correctPhoneNumber()(df(colName), df(tmpColumn)))  
42. .select(keyCol,colName, tmpColumn)  
43. df.show()  
44. 
45. // write to temp file and result  
46. val outputPath = dataPath + System.currentTimeMillis().toString  
47. df.write.parquet(outputPath)  
48. dbutils.notebook.exit(outputPath)  

1.  import org.apache.spark.sql.functions._  
2.  
3.  // read paramter  
4.  val dataPath = dbutils.widgets.get("dataPath")  
5.  val keyCol = dbutils.widgets.get("keyCol")  
6.  val colNameStr = dbutils.widgets.get("column")  
7.    
8.  var df = spark.read.parquet(dataPath)  
9.  val colNameLst = colNameStr.split("//")  
10.   
11. df= df.select(keyCol)  
12. for( w <- 0 to colNameLst.length - 1)   
13.   {   
14.     df= df.withColumn(colNameLst(w), lit(1))  
15.   }  
16. df= df.drop(keyCol)  
17. 
18. // write to temp file and result  
19. val outputPath = dataPath + (current_timestamp()).expr.eval().toString  
20. df.write.parquet(outputPath)  
21. dbutils.notebook.exit(outputPath)

1.  import org.apache.spark.sql.functions._  
2.  import org.apache.spark.sql.types.{  
3.      StructType, StructField, StringType, IntegerType}  
4.  import org.apache.spark.sql.Row  
5.  import org.apache.spark.sql.Column  
6.  
7.  
8.  // Configure common feature notebook  
9.  val downloadFileNotebook = "<path_to_download_file_notebook>"  
10. val writeFileNotebook = "<path_to_write_file_notebook>"  
11. val contraintFeaturesFolder = "<path_to_feature_notebook_folder>"  
12. 
13. val tmpFolder = "/tmp/"  
14. 
15. // These values are hard coded here, but you can update to get it from parameters  
16. val dataFile = "user.csv"  
17. val dataFileContainer = "input"    
18. val validationFile = "user-configuration.json"  
19. val validationFileContainer = "configuration"  
20. val keyCol = "User_ID"  
21.   
22.   
23. val hashString = (current_timestamp()).expr.eval().toString  
24. // Download data file  
25. val outputFileData = tmpFolder + "dataTemp" + hashString  
26. dbutils.notebook.run(downloadFileNotebook, 60, Map("containerName" -> dataFileContainer, "fileName" -> dataFile, "outputFile" -> outputFileData, "fileType" -> "csv"))  
27.   
28. //download validation file  
29. val outputFileValidation = tmpFolder + "validation" + hashString  
30. dbutils.notebook.run(downloadFileNotebook, 60, Map("containerName" -> validationFileContainer, "fileName" -> validationFile, "outputFile" -> outputFileValidation, "fileType" -> "json"))  
31.   
32. // Read validation file  
33. val constraintDF = spark.read.json(outputFileValidation)  
34.   
35. // Read data file  
36. var dataDf = spark.read.parquet(outputFileData)  
37. val dataOrginDf = spark.read.parquet(outputFileData)  
38. 
39. //Read constraint check  
40. val conDf = constraintDF.select($"action", $"param.*")  
41.   
42. val tmpColumn = "tmpCheck"  
43.   
44. //create empty failzone dataframe  
45. val schema = StructType(  
46.     StructField(keyCol, StringType, false) :: Nil)  
47.   
48. // var failDF = spark.createDataFrame(sc.emptyRDD[Row], schema)  
49. var constOutputLst = Array[String]()  
50. // execute constraint check  
51. conDf.collect().foreach(row => {  
52.   var paramMap = scala.collection.mutable.Map[String, String]()  
53.   paramMap("dataPath") = outputFileData  
54.   paramMap("keyCol") = keyCol  
55.   paramMap("tmpColumn") = tmpColumn  
56.   // create param map  
57.   for( w <- 1 to conDf.columns.size - 1)   
58.     {   
59.       if (row(w) != null) {  
60.         paramMap(conDf.columns(w)) = row(w).toString  
61.       }  
62.     }  
63.   // execute Check  
64.   val outputPath = dbutils.notebook.run(contraintFeaturesFolder + row(0).toString, 60, paramMap)   
65.   // add output file path to list for process later   
66.   constOutputLst :+= outputPath  
67. })  
68. 
69. var failDF = spark.createDataFrame(sc.emptyRDD[Row], schema)  
70.   
71. //Process constraint output file  
72. constOutputLst.foreach (file => {  
73.   val outDF = spark.read.parquet(file)  
74.   // check to remove column, if returned dataframe not have key Column => remove column  
75.   if (outDF.columns.filter(_ == keyCol).length > 0) {  
76.     // get all column but keycol and tempCheck column  
77.     outDF.columns.filter(_ != keyCol).filter(_ != tmpColumn).foreach(colName => {  
78.       // update or add new column from output file  
79.       val tmp = outDF.select(keyCol, colName)  
80.       dataDf = dataDf.drop(colName).join(tmp, dataDf(keyCol) === tmp(keyCol), "left").drop(tmp(keyCol))  
81.     })  
82.     // add fail data to fail df  
83.     failDF = failDF.union(outDF.filter(col(tmpColumn) === 0).select(keyCol))  
84.   } else {  
85.     // remove column  
86.     outDF.columns.filter(_ != tmpColumn).foreach(colName => {  
87.       dataDf = dataDf.drop(colName)  
88.     })  
89.   }  
90.     
91. })  
92. 
93. //save to correctZone  
94. val correctZoneDF = dataDf.join(failDF, dataDf(keyCol) === failDF(keyCol), "left_anti")  
95. if (correctZoneDF.collect().size > 0) {  
96.   val dataPath = outputFileData + "output" + System.currentTimeMillis().toString  
97.   val output_container_name = "output"  
98.   val outputPath = dataFile  
99.   correctZoneDF.write.parquet(dataPath)  
100.      dbutils.notebook.run(writeFileNotebook, 60, Map("dataPath" -> dataPath, "output_container_name" -> output_container_name, "outputPath" -> outputPath))  
101.    }  
102.    
103.    // Remove temp file if it exists  
104.    dbutils.fs.rm(outputFileValidation, true)  
105.    dbutils.fs.rm(outputFileData, true)  
106.    
107.    // return result check  
108.    dbutils.notebook.exit("1")