SparkSQL의 Avro 스키마 및 마루판 형식 읽기/쓰기

Question

SparkSQL에서 마루판 파일을 쓰고 읽으려고합니다. 스키마 진화의 이유로, 필자의 글과 읽기에 Avro 스키마를 사용하고 싶다.

제 생각에 이것은 Spark 외부에서 (또는 Spark 내에서 수동으로) 가능합니다. AvroParquetWriter 및 Avro의 일반 API 그러나 SparkSQL의 write() 및 read() 메서드 (DataFrameWriter 및 DataFrameReader와 함께 작동)를 사용하고 SparkSQL과 잘 통합되어 있습니다 (데이터 세트를 작성하고 읽음).

나는이 일을하는 방법을 알아낼 수 없으며, 이것이 가능한지 궁금해하고 있습니다. SparkSQL 마루판 형식이 지원하는 유일한 옵션은 "압축"및 "mergeSchema"입니다. 즉, 대체 스키마 형식이나 대체 스키마를 지정하는 옵션이 없습니다. 즉, SparkSQL API를 사용하여 Avro 스키마를 사용하여 Parquet 파일을 읽거나 쓸 수있는 방법이 없다는 것입니다. 하지만 아마도 나는 뭔가를 놓친 것일까 요?

분명히하기 위해 기본적으로 Avro 스키마를 쓰기에 대한 파르 케타 메타 데이터에 추가하고 읽기 (마루판 형식 -> Avro 스키마 -> SparkSQL 내부 형식)에 하나 더 많은 변환 계층을 추가하지만 구체적으로 누락 된 열에 대한 기본값을 추가 할 수 있습니다 (Avro 스키마는 지원하지만 파르 케 스키마는 지원하지 않음).

또한 Avro를 마루로 변환하거나 Avquet을 마루로 변환하는 방법을 찾고 있지 않으며 SparkSQL에서 일반 Avro를 읽고 쓸 수있는 방법을 찾고 있지 않습니다. databricks/spark-avro를 사용하여이 작업을 수행 할 수 있습니다.

Answer 1

나는 비슷한 것을하고있다. 나는 avro 스키마를 사용하여 마루 파일에 글을 쓰지 만, avro로 읽지는 않습니다. 그러나 동일한 기술이 읽기에서도 작동해야합니다. 이것이 최선의 방법인지 확실하지 않지만 어쨌든 여기에 있습니다 : avro 스키마가있는 AvroData.avsc가 있습니다.

KafkaUtils.createDirectStream[String,Array[Byte],StringDecoder,DefaultDecoder,Tuple2[String, Array[Byte]]](ssc, kafkaProps, fromOffsets, messageHandler) 


kafkaArr.foreachRDD { (rdd,time) 
     => { val schema = SchemaConverters.toSqlType(AvroData.getClassSchema).dataType.asInstanceOf[StructType] val ardd = rdd.mapPartitions{itr => 
       itr.map { r => 
try { 
        val cr = avroToListWithAudit(r._2, offsetSaved, loadDate, timeNow.toString) 
        Row.fromSeq(cr.toArray) 
    } catch{ 
     case e:Exception => LogHandler.log.error("Exception while converting to Avro" + e.printStackTrace()) 
     System.exit(-1) 
     Row(0) //This is just to allow compiler to accept. On exception, the application will exit before this point 
} 
} 
} 


    public static List avroToListWithAudit(byte[] kfkBytes, String kfkOffset, String loaddate, String loadtime) throws IOException { 
     AvroData av = getAvroData(kfkBytes); 
     av.setLoaddate(loaddate); 
     av.setLoadtime(loadtime); 
     av.setKafkaOffset(kfkOffset); 
     return avroToList(av); 
    } 



public static List avroToList(AvroData a) throws UnsupportedEncodingException{ 
     List<Object> l = new ArrayList<>(); 
     for (Schema.Field f : a.getSchema().getFields()) { 
      String field = f.name().toString(); 
      Object value = a.get(f.name()); 
      if (value == null) { 
       //System.out.println("Adding null"); 
       l.add(""); 
      } 
      else { 
       switch (f.schema().getType().getName()){ 
        case "union"://System.out.println("Adding union"); 
         l.add(value.toString()); 
         break; 

        default:l.add(value); 
         break; 
       } 

      } 
     } 
     return l; 
    } 

getAvroData 메서드에는 원시 바이트에서 avro 객체를 구성하는 코드가 있어야합니다. 또한 각 속성 setter를 명시 적으로 지정하지 않고도이 작업을 수행 할 수있는 방법을 찾으려고 노력하고 있지만 실제로 존재하지 않는 것처럼 보입니다.

public static AvroData getAvroData (bytes) 
{ 
AvroData av = AvroData.newBuilder().build(); 
     try { 
      av.setAttr(String.valueOf("xyz")); 
     ..... 
    } 
    } 

는