Beste gebruiker, hier is een gestructureerd kader voor Spark data kwaliteitsvalidatie afgestemd op uw vereisten: ## 1. Framework Architectuur ```scala import org.apache.spark.sql.{SparkSession, DataFrame} import org.apache.spark.sql.functions._ import io.prometheus.client.{Counter, Gauge} object DataQualityFramework { val spark: SparkSession = SparkSession.builder() .appName("DataQualityValidation") .config("spark.sql.adaptive.enabled", "true") .config("spark.sql.adaptive.coalescePartitions.enabled", "true") .getOrCreate() // Prometheus metrics val completenessErrors: Counter = Counter.build() .name("data_quality_completeness_errors") .help("Aantal completeness validatiefouten") .register() val uniquenessErrors: Counter = Counter.build() .name("data_quality_uniqueness_errors") .help("Aantal uniqueness validatiefouten") .register() } ``` ## 2. Data Ingestie & Preprocessing ```scala class KafkaJSONProcessor { def readKafkaStream(jsonPath: String): DataFrame = { DataQualityFramework.spark.read .option("multiLine", true) .option("mode", "PERMISSIVE") .json(jsonPath) .withColumn("ingestion_timestamp", current_timestamp()) } } ``` ## 3. Volledigheid (Completeness) Checks ```scala trait CompletenessChecks { def checkCompleteness(df: DataFrame, mandatoryColumns: Seq[String]): Map[String, Long] = { import DataQualityFramework.spark.implicits._ mandatoryColumns.map { column => val nullCount = df.filter(col(column).isNull).count() if (nullCount > 0) { DataQualityFramework.completenessErrors.inc(nullCount) } column -> nullCount }.toMap } def calculateCompletenessScore(df: DataFrame, columns: Seq[String]): Double = { val totalRecords = df.count() val nullCounts = columns.map(col => df.filter(col(col).isNull).count()).sum (totalRecords - nullCounts).toDouble / totalRecords * 100 } } ``` ## 4. Uniekheid (Uniqueness) Checks ```scala trait UniquenessChecks { def checkUniqueConstraints(df: DataFrame, uniqueColumns: Seq[String]): Map[String, Long] = { import DataQualityFramework.spark.implicits._ uniqueColumns.map { column => val duplicateCount = df.groupBy(col(column)) .count() .filter(col("count") > 1) .count() if (duplicateCount > 0) { DataQualityFramework.uniquenessErrors.inc(duplicateCount) } column -> duplicateCount }.toMap } def identifyDuplicateRecords(df: DataFrame, keyColumns: Seq[String]): DataFrame = { val windowSpec = Window.partitionBy(keyColumns.map(col): _*) .orderBy(col("ingestion_timestamp").desc) df.withColumn("row_num", row_number().over(windowSpec)) .filter(col("row_num") > 1) .drop("row_num") } } ``` ## 5. Geïntegreerde Validatie Pipeline ```scala class DataQualityValidator extends CompletenessChecks with UniquenessChecks { def validateDataQuality(df: DataFrame, config: ValidationConfig): ValidationResult = { // Volledigheid checks val completenessResults = checkCompleteness(df, config.mandatoryColumns) val completenessScore = calculateCompletenessScore(df, config.mandatoryColumns) // Uniekheid checks val uniquenessResults = checkUniqueConstraints(df, config.uniqueColumns) // Genereer rapport ValidationResult( completenessResults = completenessResults, completenessScore = completenessScore, uniquenessResults = uniquenessResults, totalRecords = df.count(), validationTimestamp = java.time.Instant.now() ) } def processHourlyBatch(jsonPath: String, config: ValidationConfig): Unit = { val processor = new KafkaJSONProcessor val df = processor.readKafkaStream(jsonPath) val result = validateDataQuality(df, config) // Export metrics naar Prometheus exportMetricsToPrometheus(result) // Genereer Grafana dashboard data generateGrafanaMetrics(result) // Log resultaten logValidationResults(result) } } ``` ## 6. Configuratie Management ```scala case class ValidationConfig( mandatoryColumns: Seq[String], uniqueColumns: Seq[String], validationThreshold: Double = 95.0, alertThreshold: Double = 90.0 ) case class ValidationResult( completenessResults: Map[String, Long], completenessScore: Double, uniquenessResults: Map[String, Long], totalRecords: Long, validationTimestamp: java.time.Instant ) ``` ## 7. Prometheus & Grafana Integratie ```scala trait MonitoringIntegration { def exportMetricsToPrometheus(result: ValidationResult): Unit = { // Prometheus metrics export val completenessGauge = Gauge.build() .name("data_quality_completeness_score") .help("Volledigheid score percentage") .register() completenessGauge.set(result.completenessScore) } def generateGrafanaMetrics(result: ValidationResult): Unit = { // JSON format voor Grafana val grafanaData = s""" { "completeness_score": ${result.completenessScore}, "total_records": ${result.totalRecords}, "timestamp": "${result.validationTimestamp}" } """ // Integreer met Grafana HTTP API } def triggerAlerts(result: ValidationResult, config: ValidationConfig): Unit = { if (result.completenessScore < config.alertThreshold) { // Trigger alert via Prometheus Alertmanager } } } ``` ## 8. Implementatie Voorbeeld ```scala object HourlyBatchProcessor extends App with MonitoringIntegration { val validator = new DataQualityValidator val config = ValidationConfig( mandatoryColumns = Seq("id", "timestamp", "event_type"), uniqueColumns = Seq("id"), alertThreshold = 90.0 ) // Schedule hourly processing val jsonPath = "/path/to/kafka/json/hourly_batch/" validator.processHourlyBatch(jsonPath, config) // Error handling en retry logic def withRetry[T](operation: => T, maxRetries: Int = 3): T = { // Implementatie retry mechanisme } } ``` ## 9. Performance Optimalisaties ```scala object OptimizationTips { // Gebruik Delta Lake voor betere prestaties def enableDeltaOptimizations(): Unit = { DataQualityFramework.spark.conf.set("spark.sql.adaptive.enabled", "true") DataQualityFramework.spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true") } // Cache strategie voor grote datasets def smartCache(df: DataFrame): DataFrame = { if (df.count() > 1000000) { df.cache() } else { df } } } ``` ## 10. Uitrol en Monitoring 1. **Deployment**: Package als JAR en deploy op Spark cluster 2. **Monitoring**: Configureer Prometheus scraping 3. **Alerting**: Stel drempelwaarden in voor automatische waarschuwingen 4. **Dashboard**: Bouw Grafana dashboard voor real-time monitoring Dit framework biedt schaalbare, monitorbare data kwaliteitsvalidatie specifiek voor uw use case met uurelijkse batches van miljoen records.