Feature/etl optimization

admin-app/src/main/resources/application.conf

@@ -94,5 +94,11 @@ drill {
         fetchSize = ${?DRILL_ETL_FETCH_SIZE}
         batchSize = 1000
         batchSize = ${?DRILL_ETL_BATCH_SIZE}
+        extractionLimit = 1000000
+        extractionLimit = ${?DRILL_ETL_EXTRACTION_LIMIT}
+        transformationBufferSize = 2000
+        transformationBufferSize = ${?DRILL_ETL_TRANSFORMATION_BUFFER_SIZE}
+        loggingFrequency = 10
+        loggingFrequency = ${?DRILL_ETL_LOGGING_FREQUENCY}
     }
 }

admin-etl/src/main/kotlin/com/epam/drill/admin/etl/DataExtractor.kt

@@ -30,14 +30,14 @@ import java.time.Instant
  *     - how data will be transformed,
  *     - how metrics will be stored.
  */
-interface DataExtractor<T> {
+interface DataExtractor<T: EtlRow> {
     val name: String
     suspend fun extract(
         groupId: String,
         sinceTimestamp: Instant,
         untilTimestamp: Instant,
         emitter: FlowCollector<T>,
-        onExtractCompleted: suspend (EtlExtractingResult) -> Unit
+        onExtractingProgress: suspend (EtlExtractingResult) -> Unit
     )
 }
 

admin-etl/src/main/kotlin/com/epam/drill/admin/etl/DataLoader.kt

@@ -29,14 +29,15 @@ import java.time.Instant
  * - Uses `batchSize` to control transaction size and DB pressure.
  * - Is resilient to partial failures
  */
-interface DataLoader<T> {
+interface DataLoader<T: EtlRow> {
     val name: String
     suspend fun load(
         groupId: String,
         sinceTimestamp: Instant,
         untilTimestamp: Instant,
         collector: Flow<T>,
-        onLoadCompleted: suspend (EtlLoadingResult) -> Unit
+        onLoadingProgress: suspend (EtlLoadingResult) -> Unit = {},
+        onStatusChanged: suspend (EtlStatus) -> Unit = {},
     ): EtlLoadingResult
 
     suspend fun deleteAll(groupId: String)

admin-etl/src/main/kotlin/com/epam/drill/admin/etl/DataTransformer.kt

@@ -0,0 +1,36 @@
+/**
+ * Copyright 2020 - 2022 EPAM Systems
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.epam.drill.admin.etl
+
+import kotlinx.coroutines.flow.Flow
+
+interface DataTransformer<in T: EtlRow, out R: EtlRow> {
+    val name: String
+    suspend fun transform(
+        groupId: String,
+        collector: Flow<T>,
+    ): Flow<R>
+}
+
+class NopTransformer<T: EtlRow> : DataTransformer<T, T> {
+    override val name: String = "nop-transformer"
+    override suspend fun transform(
+        groupId: String,
+        collector: Flow<T>,
+    ): Flow<T> = collector
+}
+
+val untypedNopTransformer = NopTransformer<UntypedRow>()

admin-etl/src/main/kotlin/com/epam/drill/admin/etl/EtlExtractingResult.kt

@@ -16,7 +16,6 @@
 package com.epam.drill.admin.etl
 
 class EtlExtractingResult(
-    val success: Boolean,
-    val duration: Long,
+    val duration: Long = 0,
     val errorMessage: String? = null
 )

admin-etl/src/main/kotlin/com/epam/drill/admin/etl/EtlLoadingResult.kt

@@ -20,17 +20,15 @@ import java.time.Instant
 data class EtlLoadingResult(
     val lastProcessedAt: Instant,
     val processedRows: Long = 0L,
-    val status: EtlStatus,
     val duration: Long? = null,
     val errorMessage: String? = null
 ) : Comparable<EtlLoadingResult> {
     val isFailed
-        get() = status == EtlStatus.FAILED
+        get() = errorMessage != null
 
     operator fun plus(other: EtlLoadingResult): EtlLoadingResult {
         val failed = this.isFailed || other.isFailed
         return EtlLoadingResult(
-            status = if (!failed) other.status else EtlStatus.FAILED,
             lastProcessedAt = if (!failed) other.lastProcessedAt else this.lastProcessedAt,
             processedRows = this.processedRows + other.processedRows,
             duration = listOfNotNull(this.duration, other.duration).sum(),

admin-etl/src/main/kotlin/com/epam/drill/admin/etl/EtlMetadata.kt

@@ -24,9 +24,10 @@ data class EtlMetadata(
     val loaderName: String,
     val lastProcessedAt: Instant,
     val lastRunAt: Instant,
-    val lastDuration: Long,
-    val lastRowsProcessed: Long,
+    val lastLoadDuration: Long = 0L,
+    val lastExtractDuration: Long = 0L,
+    val lastRowsProcessed: Long = 0L,
     val status: EtlStatus,
-    val errorMessage: String?,
+    val errorMessage: String? = null,
 )
 

admin-etl/src/main/kotlin/com/epam/drill/admin/etl/EtlMetadataRepository.kt

@@ -15,13 +15,35 @@
  */
 package com.epam.drill.admin.etl
 
+import java.time.Instant
+
 interface EtlMetadataRepository {
     suspend fun getAllMetadata(groupId: String): List<EtlMetadata>
-    suspend fun getAllMetadataByExtractor(groupId: String, pipelineName: String, extractorName: String): List<EtlMetadata>
-    suspend fun getMetadata(groupId: String, pipelineName: String, extractorName: String, loaderName: String): EtlMetadata?
+    suspend fun getAllMetadataByExtractor(
+        groupId: String,
+        pipelineName: String,
+        extractorName: String
+    ): List<EtlMetadata>
+
+    suspend fun getMetadata(
+        groupId: String,
+        pipelineName: String,
+        extractorName: String,
+        loaderName: String
+    ): EtlMetadata?
+
     suspend fun saveMetadata(metadata: EtlMetadata)
-    suspend fun accumulateMetadata(metadata: EtlMetadata)
+    suspend fun accumulateMetadataByLoader(
+        groupId: String, pipelineName: String, extractorName: String, loaderName: String,
+        lastProcessedAt: Instant? = null,
+        status: EtlStatus? = null, loadDuration: Long = 0L, rowsProcessed: Long = 0L,
+        errorMessage: String? = null
+    )
     suspend fun deleteMetadataByPipeline(groupId: String, pipelineName: String)
-    suspend fun accumulateMetadataDurationByExtractor(groupId: String, pipelineName: String, extractorName: String, duration: Long)
+    suspend fun accumulateMetadataByExtractor(
+        groupId: String, pipelineName: String, extractorName: String,
+        status: EtlStatus? = null, extractDuration: Long = 0L,
+        errorMessage: String? = null
+    )
 }
 

admin-etl/src/main/kotlin/com/epam/drill/admin/etl/EtlPipeline.kt

@@ -35,16 +35,17 @@ import java.time.Instant
  *     - failures in loaders or extractor are propagated to the pipeline,
  *     - pipeline may cancel child coroutines and report status to `EtlOrchestrator`.
  */
-interface EtlPipeline<T> {
+interface EtlPipeline<in T : EtlRow, out R : EtlRow> {
     val name: String
-    val extractor: DataExtractor<T>
-    val loaders: List<DataLoader<T>>
+    val extractor: DataExtractor<in T>
+    val loaders: List<Pair<DataTransformer<T, R>, DataLoader<out R>>>
     suspend fun execute(
         groupId: String,
         sinceTimestampPerLoader: Map<String, Instant>,
         untilTimestamp: Instant,
-        onExtractCompleted: suspend (EtlExtractingResult) -> Unit = {},
-        onLoadCompleted: suspend (loaderName: String, result: EtlLoadingResult) -> Unit = { _, _ -> },
+        onExtractingProgress: suspend (EtlExtractingResult) -> Unit = {},
+        onLoadingProgress: suspend (loaderName: String, result: EtlLoadingResult) -> Unit = { _, _ -> },
+        onStatusChanged: suspend (loaderName: String, status: EtlStatus) -> Unit = { _, _ -> },
     ): EtlProcessingResult
 
     suspend fun cleanUp(groupId: String)

admin-etl/src/main/kotlin/com/epam/drill/admin/etl/EtlRow.kt

@@ -0,0 +1,22 @@
+/**
+ * Copyright 2020 - 2022 EPAM Systems
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.epam.drill.admin.etl
+
+import java.time.Instant
+
+open class EtlRow(
+    val timestamp: Instant,
+)

admin-etl/src/main/kotlin/com/epam/drill/admin/etl/UntypedRow.kt

@@ -0,0 +1,23 @@
+/**
+ * Copyright 2020 - 2022 EPAM Systems
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.epam.drill.admin.etl
+
+import java.time.Instant
+
+class UntypedRow(
+    timestamp: Instant,
+    map: Map<String, Any?>
+): EtlRow(timestamp), Map<String, Any?> by HashMap(map)

admin-etl/src/main/kotlin/com/epam/drill/admin/etl/flow/LruMap.kt

@@ -0,0 +1,55 @@
+/**
+ * Copyright 2020 - 2022 EPAM Systems
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.epam.drill.admin.etl.flow
+
+/**
+ * A simple LRU (Least Recently Used) map implementation that evicts the oldest entry
+ * when the maximum size is reached.
+ */
+class LruMap<K, V>(
+    private val maxSize: Int,
+    private val onEvict: (K, V) -> Unit
+) {
+    private val map = LinkedHashMap<K, V>(16, 0.75f, true)
+
+    val size: Int
+        get() = map.size
+
+    fun compute(key: K, update: (V?) -> V) {
+        map[key] = update(map[key])
+        if (map.size >= maxSize) {
+            evictOldest()
+        }
+    }
+
+    fun evictOldest() {
+        val it = map.entries.iterator()
+        if (!it.hasNext()) return
+
+        val entry = it.next()
+        it.remove()
+        onEvict(entry.key, entry.value)
+    }
+
+    fun evictAll() {
+        val it = map.entries.iterator()
+        while (it.hasNext()) {
+            val entry = it.next()
+            it.remove()
+            onEvict(entry.key, entry.value)
+        }
+    }
+}