0

有没有一种方法可以在不使用 Apache Spark 的情况下将 scv 文件转换为 Avro。我看到大多数帖子都建议使用 spark ，但在我的情况下我不能。我在单独的文件中有一个架构。我正在考虑一些自定义序列化器和反序列化器，它们将使用 Schema 并将 csv 转换为 avro。任何类型的参考都对我有用。谢谢

4

2 回答 2

1

如果你只有字符串和原语，你可以很容易地把这样一个粗略的实现放在一起：

def csvToAvro(file: Sting, schema: Schema) = {
  val rec = new GenericData.Record(schema)
  val types = schema
    .getFields
    .map { f => f.pos -> f.schema.getType }

  Source.fromFile(file)
   .getLines
   .map(_.split("_").toSeq)
   .foreach { data => 
     (data zip types)
       .foreach {
         case (str, (idx, STRING)) => rec.put(idx, str)
         case (str, (idx, INT)) => rec.put(idx, str.toInt)
         case (str, (idx, LONG)) => rec.put(idx, str.toLong)
         case (str, (idx, FLOAT)) => rec.put(idx, str.toFloat)
         case (str, (idx, DOUBLE)) => rec.put(idx, str.toDouble)
         case (str, (idx, BOOLEAN)) => rec.put(idx, str.toBoolean)  
         case (str, (idx, unknown)) => throw new IllegalArgumentException(s"Don't know how to convert $str to $unknown at $idx))
       }
  }
  rec
}

请注意，这不处理可为空的字段：对于那些类型将是 的字段UNION，您必须查看架构内部以找出实际的数据类型。

此外，“解析 csv”在这里非常粗略（仅在逗号处拆分并不是一个好主意，因为如果字符串字段恰好包含,在数据中，或者如果字段用双引号转义，它将中断） .

此外，您可能需要添加一些完整性检查以确保例如 csv 行中的字段数与架构中的字段数等相匹配。

但是尽管有上述考虑，这应该足以说明方法并让您入门。

于 2017-07-07T15:07:44.090 回答
0

Avro 是一种开放格式，有许多语言支持它。

只需选择一个，例如 python 也支持csv。但是 Go 可以，Java 也可以。

于 2017-07-07T14:50:11.713 回答