feature/Option-to-import-JSON-as-is #18
Conversation
There was a problem hiding this comment.
Hello @ilikutle,
Thanks a lot for working on this issue and sending the pull request! We really appreciate it!
I have submitted some request change suggestions. Please have a look. Let me know if you have questions or want to discuss them.
Could you please also add a small test for this? You can check out how we use embedded-kafka library on KafkaTopicDataImportIT class. You can create another similar class (KafkaTopicDataImporterAsJsonIT) with a single test, and update the user properties accordingly.
doc/user_guide/user_guide.md
Outdated
| The first two columns are used to store the metadata about Kafka topic partition | ||
| ```sql | ||
| CREATE OR REPLACE TABLE <schema_name>.<table_name> ( | ||
| -- These columns match the Kafka topic schema |
There was a problem hiding this comment.
| -- These columns match the Kafka topic schema | |
| -- Single column as JSON string for Kafka topic record |
| @@ -49,6 +49,7 @@ object KafkaTopicDataImporter extends LazyLogging { | |||
| val maxRecords = kafkaProperties.getMaxRecordsPerRun() | |||
| val minRecords = kafkaProperties.getMinRecordsPerRun() | |||
| val timeout = kafkaProperties.getPollTimeoutMs() | |||
| val singleColJson = kafkaProperties.getSingleColJson() | |||
There was a problem hiding this comment.
| val singleColJson = kafkaProperties.getSingleColJson() | |
| val singleColumnJson = kafkaProperties.getSingleColJson() |
| val avroRow = AvroRow(record.value()).getValues().map(_.asInstanceOf[AnyRef]) | ||
| val exasolRow: Seq[Object] = avroRow ++ metadata | ||
| iterator.emit(exasolRow: _*) | ||
| } |
There was a problem hiding this comment.
Could you please add one private function for a row?
Suggestion:
val exasolRow: Seq[Object] = getAvroRow(singleColumnJson) ++ metadata
iterator.emit(exasolRow: _*)where the getAvroRow function contains the if else part.
Co-authored-by: Muhammet Orazov <[email protected]>
#16 fix