AbsaOSS / AbsaOSS/ABRiS

Getting zero byte files using abris

Đang mở
#307 3 bình luận 0 reaction 0 người được giao Xem trên GitHub
Ngôn ngữ chính
Scala
Star
242
Fork
84
Merge trung bình
10 giờ 12 phút
Pull request đã merge (30 ngày)
2

Mô tả

I was using abris with confluent schema registry to deserialize avro records received from kafka source.
When i use confluent schema registry and manually get the schema and pass it to spark default from_avro function by skipping first 6 bytes i was able to read records. I wish to do the same using abris. Since abris as a library does that. But when i am using abris 0 byte files are written. Is this issue with Abris ?

Working code without Abris:

```
package pruthvi.kafka.poc
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.avro.functions.from_avro
import org.apache.spark.sql.functions.col
import org.apache.spark.sql.streaming.Trigger
import io.confluent.kafka.schemaregistry.client.rest.RestService
import org.apache.avro.Schema

import java.util
import org.apache.spark.sql.avro.functions._

object Important1 {

def main(args: Array[String]): Unit = {
println("Hello world!")

try {
val spark: SparkSession = SparkSession.builder
.master("local[3]")
.appName("Kafka testing")
.config("spark.streaming.stopGracefullyOnShutdown", "true")
.config("spark.sql.shuffle.partitions",3)
.getOrCreate()
spark.conf.set("spark.sql.avro.compression.codec", "uncompressed")

val topicName = "foo"

val df = spark.readStream
.format("kafka")
.option("kafka.security.protocol", "SSL")
.option("kafka.ssl.truststore.location", "certs/truststore.jks")
.option("kafka.ssl.keystore.location", "certs/keystore.jks")
.option("kafka.ssl.key.password", "foo")
.option("kafka.ssl.keystore.password", "foo")
.option("kafka.ssl.truststore.password", "foo")
.option("kafka.bootstrap.servers", "x:16501,y:16501,z:16501")
.option("subscribe", topicName)
.option("kafka.group.id", "foo")
.option("startingOffsets", "earliest")
.load()

val schemaRegistryURL = "url"
val restService = new RestService(schemaRegistryURL)
val valueRestResponseSchema = restService.getLatestVersion(topicName)
val jsonSchema = valueRestResponseSchema.getSchema

import spark.implicits._
val dsAvroRecord = df
.selectExpr("substring(value, 6) as avro_value")
.select(
from_avro($"avro_value", jsonSchema, fromAvroConfig).as("RecordValue"))

dsAvroRecord.writeStream
.outputMode("append")
.format("json")
.option("path", "output")
.trigger(Trigger.ProcessingTime(1))
.option("checkpointLocation", "chk_point_dir")
.start().awaitTermination()
}
catch{
case e:Exception=>{
println(e.printStackTrace())
}
}
}
}
```
With Abris :
```
package pruthvi.kafka.poc
package examples

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.col
import org.apache.spark.sql.streaming.Trigger
import za.co.absa.abris.config.AbrisConfig

import scala.concurrent.duration.Duration

object readStreamingData {

def main(args: Array[String]): Unit = {

val spark: SparkSession = SparkSession.builder
.master("local[4]")
.appName("Kafka testing")
.config("spark.streaming.stopGracefullyOnShutdown", "true")
.getOrCreate()

val topicName = "foo"
val df = spark.readStream
.format("kafka")
.option("kafka.security.protocol", "SSL")
.option("kafka.ssl.truststore.location", "certs/truststore.jks")
.option("kafka.ssl.keystore.location", "certs/keystore.jks")
.option("kafka.ssl.key.password", "")
.option("kafka.ssl.keystore.password", "")
.option("kafka.ssl.truststore.password", "")
.option(
"kafka.bootstrap.servers",
"x:16501,y:16501,z:16501"
)
.option("kafka.group.id", "foo")
.option("subscribe", topicName)
.option("startingOffsets", "earliest")
.option("maxOffsetsPerTrigger","50")
.load()

val abrisConfig =
AbrisConfig.fromConfluentAvro.downloadReaderSchemaByLatestVersion
.andTopicNameStrategy(topicName)
.usingSchemaRegistry("url")

import za.co.absa.abris.avro.functions.from_avro
val deserialized = df.select(from_avro(col("value"), abrisConfig) as 'data)

deserialized
.writeStream
.option("path", "output")
.option("checkpointLocation", "chk_point_dir")
.outputMode("append")
.format("parquet")
.trigger(Trigger.ProcessingTime(1000))
.start()
.awaitTermination()

}
}
```

Hướng dẫn đóng góp

Chưa lập chỉ mục được hướng dẫn đóng góp cho kho mã nguồn này

Đánh giá

Issue này chưa được đánh giá.

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.