Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

#137: Make column non nullable #140

Merged
merged 2 commits into from
Jun 18, 2020
Merged

#137: Make column non nullable #140

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b609f68
#137: Make column non nullable
kevinwallimann Jun 15, 2020
c2a0bf2
Merge branch 'develop' into feature/137-avro-non-nullable2
kevinwallimann Jun 18, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions driver/src/test/scala/za/co/absa/hyperdrive/driver/drivers/KafkaToKafkaDockerTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ import java.time.Duration
import java.util.UUID.randomUUID
import java.util.{Collections, Properties}

import org.apache.avro.Schema.Parser
import org.apache.avro.Schema.{Parser, Type}
import org.apache.avro.generic.{GenericData, GenericRecord}
import org.apache.avro.util.Utf8
import org.apache.hadoop.fs.{FileSystem, Path}
Expand Down Expand Up @@ -59,7 +59,7 @@ class KafkaToKafkaDockerTest extends FlatSpec with Matchers with SparkTestBase w

val valueSchemaString = raw"""{"type": "record", "name": "$sourceTopic", "fields": [
{"type": "int", "name": "some_id"},
{"type": "string", "name": "value_field"}
{"type": ["string", "null"], "name": "value_field"}
]}"""
val valueSchema = new Parser().parse(valueSchemaString)

Expand Down Expand Up @@ -138,6 +138,10 @@ class KafkaToKafkaDockerTest extends FlatSpec with Matchers with SparkTestBase w
keyFieldNames should contain theSameElementsAs List("some_id", "key_field")
records.map(_.key().get("some_id")) should contain theSameElementsInOrderAs List.tabulate(numberOfRecords)(_ / 5)
records.map(_.key().get("key_field")).distinct should contain theSameElementsAs List(new Utf8("keyHello"))

records.head.value().getSchema.getField("some_id").schema().getType shouldBe Type.INT
records.head.value().getSchema.getField("value_field").schema().getTypes
.asScala.map(_.getType) should contain theSameElementsAs Seq(Type.STRING, Type.NULL)

val valueFieldNames = records.head.value().getSchema.getFields.asScala.map(_.name())
valueFieldNames should contain theSameElementsAs List("some_id", "value_field")
Expand Down
18 changes: 13 additions & 5 deletions ...rive/ingestor/implementation/decoder/avro/confluent/ConfluentAvroKafkaStreamDecoder.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,18 +18,19 @@ package za.co.absa.hyperdrive.ingestor.implementation.decoder.avro.confluent
import org.apache.commons.configuration2.Configuration
import org.apache.commons.lang3.{RandomStringUtils, StringUtils}
import org.apache.logging.log4j.LogManager
import org.apache.spark.sql.catalyst.expressions.objects.AssertNotNull
import org.apache.spark.sql.functions.col
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.streaming.DataStreamReader
import org.apache.spark.sql.{Column, DataFrame}
import za.co.absa.abris.avro.functions.from_confluent_avro
import za.co.absa.abris.avro.read.confluent.SchemaManager._
import za.co.absa.hyperdrive.ingestor.api.context.HyperdriveContext
import za.co.absa.hyperdrive.ingestor.api.decoder.{StreamDecoder, StreamDecoderFactory}
import za.co.absa.hyperdrive.ingestor.implementation.utils.{SchemaRegistryConsumerConfigKeys, SchemaRegistrySettingsUtil}
import za.co.absa.hyperdrive.shared.configurations.ConfigurationsKeys.AvroKafkaStreamDecoderKeys._
import za.co.absa.hyperdrive.ingestor.api.utils.ConfigUtils
import za.co.absa.hyperdrive.ingestor.api.utils.ConfigUtils.getOrThrow
import za.co.absa.hyperdrive.ingestor.implementation.HyperdriveContextKeys
import za.co.absa.hyperdrive.ingestor.implementation.utils.{SchemaRegistryConsumerConfigKeys, SchemaRegistrySettingsUtil}
import za.co.absa.hyperdrive.shared.configurations.ConfigurationsKeys.AvroKafkaStreamDecoderKeys._

private[decoder] class ConfluentAvroKafkaStreamDecoder(
val topic: String,
Expand Down Expand Up @@ -60,9 +61,10 @@ private[decoder] class ConfluentAvroKafkaStreamDecoder(

private def getKeyValueDataFrame(dataFrame: DataFrame,
keySchemaRegistrySettings: Map[String, String]) = {
val keyValueDf = dataFrame.select(
val decodedDf = dataFrame.select(
from_confluent_avro(col("key"), keySchemaRegistrySettings) as 'key,
from_confluent_avro(col("value"), valueSchemaRegistrySettings) as 'value)
val keyValueDf = setColumnNonNullable(decodedDf, "value")

val keyColumnNames = keyValueDf.select("key.*").columns.toSeq
val valueColumnNames = keyValueDf.select("value.*").columns.toSeq
Expand All @@ -77,11 +79,17 @@ private[decoder] class ConfluentAvroKafkaStreamDecoder(
}

private def getValueDataFrame(dataFrame: DataFrame) = {
dataFrame
val decodedDf = dataFrame
.select(from_confluent_avro(col("value"), valueSchemaRegistrySettings) as 'data)
setColumnNonNullable(decodedDf, "data")
.select("data.*")
}

private def setColumnNonNullable(dataFrame: DataFrame, columnName: String) = {
dataFrame
.filter(col(columnName).isNotNull)
.withColumn(columnName, new Column(AssertNotNull(col(columnName).expr)))
}

}

Expand Down