在 Java 或 Scala 中将 CSV 转换为 Avro 文件答案

【问题标题】：Convert CSV to Avro file in Java or scala在 Java 或 Scala 中将 CSV 转换为 Avro 文件
【发布时间】：2018-01-14 10:55:05
【问题描述】：

是否有任何库可以将 CSV 转换为 Java 或 scala 中的 Avro 文件。

我试图用谷歌搜索它，但找不到任何库。

【问题讨论】：

CSV 模式只是命名字段，而 Avro 也有类型。您希望如何生成 Avro 方案的类型？或者你有一个在 Java/Scala 中用类表示的已知方案？
假设您可能知道架构...github.com/sspinc/csv2avro 如果不是，我认为@Bala 的答案应该是正确的
这也可能对github.com/uwegeercken/csv2avro有帮助

标签： java scala csv avro

【解决方案1】：

通过谷歌搜索，我找到了这篇文章：https://dzone.com/articles/convert-csv-data-avro-data

引用：

要使用 Hive 将 csv 数据转换为 Avro 数据，我们需要按照以下步骤操作：

创建一个存储为文本文件的 Hive 表，并指定您的 csv 分隔符。
使用“加载数据”命令将 csv 文件加载到上表。
使用 AvroSerDe 创建另一个 Hive 表。
使用“insert overwrite”命令将数据从以前的表插入到新的 Avro Hive 表中。

示例：使用 csv（student_id、subject_id、grade）

--1. Create a Hive table stored as textfile
USE test;
CREATE TABLE csv_table (
student_id INT,
subject_id INT,
marks INT)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' 
STORED AS TEXTFILE;

--2. Load csv_table with student.csv data
LOAD DATA LOCAL INPATH "/path/to/student.csv" OVERWRITE INTO TABLE test.csv_table;

--3. Create another Hive table using AvroSerDe
CREATE TABLE avro_table
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.avro.AvroSerDe'
STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerInputFormat'
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerOutputFormat'
TBLPROPERTIES (
    'avro.schema.literal'='{
      "namespace": "com.rishav.avro",
      "name": "student_marks",
      "type": "record",
      "fields": [ { "name":"student_id","type":"int"}, { "name":"subject_id","type":"int"}, { "name":"marks","type":"int"}]
    }');

--4. Load avro_table with data from csv_table
INSERT OVERWRITE TABLE avro_table SELECT student_id, subject_id, marks FROM csv_table;

【讨论】：

但它使用 Hive 进行转换。我想要通用的，所以我们可以在任何地方使用

【解决方案2】：

您可以尝试这种方式（Spark 1.6）。

people.csv

Michael, 29
Andy, 30
Justin, 19

派斯帕克

file = sc.textFile("people.csv")
df = file.map(lambda line: line.split(',')).toDF(['name','age'])

>>> df.show()
+-------+---+
|   name|age|
+-------+---+
|Michael| 29|
|   Andy| 30|
| Justin| 19|
+-------+---+

df.write.format("com.databricks.spark.avro").save("peopleavro")

人脉

{u'age': u' 29', u'name': u'Michael'}
{u'age': u' 30', u'name': u'Andy'}
{u'age': u' 19', u'name': u'Justin'}

如果你需要维护数据类型，然后创建一个模式并传递它。

schema = StructType([StructField("name",StringType(),True),StructField("age",IntegerType(),True)])

df = file.map(lambda line: line.split(',')).toDF(schema)
>>> df.printSchema()
root
 |-- name: string (nullable = true)
 |-- age: integer (nullable = true)

现在你的 avro 有

{
  "type" : "record",
  "name" : "topLevelRecord",
  "fields" : [ {
    "name" : "name",
    "type" : [ "string", "null" ]
  }, {
    "name" : "age",
    "type" : [ "int", "null" ]
  } ]
}

【讨论】：

【解决方案3】：

您可以为此目的使用 spark 或 spark-shell（带有选项：--packages org.apache.spark:spark-avro...），如果它是临时使用。

示例代码示例：

val df = spark.read.csv("example.csv") df.write.format("com.databricks.spark.avro").save("example.avro")

【讨论】：

【解决方案4】：

您可以通过以下方式轻松完成：

使用 OpenCSV 库解析您的 CSV (http://opencsv.sourceforge.net)
使用 json-smart 库 (https://github.com/netplex/json-smart-v2) 输出 Avro JSON（架构为https://en.wikipedia.org/wiki/Apache_Avro）

【讨论】：