来自 Java 中 List<String> 的数据框

Posted 2023-04-17

【中文标题】来自 Java 中 List<String> 的数据框

【英文标题】：Dataframe from List<String> in Java

【发布时间】：2017-04-26 12:04:45

【问题描述】：

Spark 版本：1.6.2 Java 版本：7

我有一个List<String> 数据。类似的东西：

[[dev, engg, 10000], [karthik, engg, 20000]..]

我知道此数据的架构。

name (String)
degree (String)
salary (Integer)

---

我试过了：

JavaRDD<String> data = new JavaSparkContext(sc).parallelize(datas);
DataFrame df = sqlContext.read().json(data);
df.printSchema();
df.show(false);

输出：

root
 |-- _corrupt_record: string (nullable = true)


+-----------------------------+
|_corrupt_record              |
+-----------------------------+
|[dev, engg, 10000]           |
|[karthik, engg, 20000]       |
+-----------------------------+

因为List<String> 不是正确的 JSON。

我需要创建一个正确的 JSON 还是有其他方法可以做到这一点？

【问题讨论】：

你为什么不创建包含这些属性的 Java bean 类，然后你可以拥有 ArrayList 然后你可以使用 sqlContext.createDataFrame(List> data, Class > beanClass)。

@abaghel 不可能为每组数据创建 Java bean 类。

【参考方案1】：

您可以从List<String> 创建DataFrame，然后使用selectExpr 和split 来获取所需的DataFrame。

public class SparkSample
public static void main(String[] args) 
    SparkConf conf = new SparkConf().setAppName("SparkSample").setMaster("local[*]");
    JavaSparkContext jsc = new JavaSparkContext(conf);
    SQLContext sqc = new SQLContext(jsc);
    // sample data
    List<String> data = new ArrayList<String>();
    data.add("dev, engg, 10000");
    data.add("karthik, engg, 20000");
    // DataFrame
    DataFrame df = sqc.createDataset(data, Encoders.STRING()).toDF();
    df.printSchema();
    df.show();
    // Convert
    DataFrame df1 = df.selectExpr("split(value, ',')[0] as name", "split(value, ',')[1] as degree","split(value, ',')[2] as salary");
    df1.printSchema();
    df1.show(); 
   

你会得到下面的输出。

root
 |-- value: string (nullable = true)

+--------------------+
|               value|
+--------------------+
|    dev, engg, 10000|
|karthik, engg, 20000|
+--------------------+

root
 |-- name: string (nullable = true)
 |-- degree: string (nullable = true)
 |-- salary: string (nullable = true)

+-------+------+------+
|   name|degree|salary|
+-------+------+------+
|    dev|  engg| 10000|
|karthik|  engg| 20000|
+-------+------+------+

您提供的示例数据有空格。如果您想删除空格并将工资类型设置为“整数”，那么您可以使用trim 和cast 函数，如下所示。

df1 = df1.select(trim(col("name")).as("name"),trim(col("degree")).‌​as("degree"),trim(co‌​l("salary")).cast("i‌​nteger").as("salary"‌​)); 

【参考方案2】：

DataFrame createNGramDataFrame(JavaRDD<String> lines) 
 JavaRDD<Row> rows = lines.map(new Function<String, Row>()
    private static final long serialVersionUID = -4332903997027358601L;

    @Override
    public Row call(String line) throws Exception 
        return RowFactory.create(line.split("\\s+"));
    
 );
 StructType schema = new StructType(new StructField[] 
        new StructField("words",
                DataTypes.createArrayType(DataTypes.StringType), false,
                Metadata.empty()) );
 DataFrame wordDF = new SQLContext(jsc).createDataFrame(rows, schema);
 // build a bigram language model
 NGram transformer = new NGram().setInputCol("words")
        .setOutputCol("ngrams").setN(2);
 DataFrame ngramDF = transformer.transform(wordDF);
 ngramDF.show(10, false);
 return ngramDF;

【参考方案3】：

在 Scala 上可以在没有 JSON 的情况下完成任务：

val data = List("dev, engg, 10000", "karthik, engg, 20000")
val intialRdd = sparkContext.parallelize(data)
val splittedRDD = intialRdd.map(current => 
  val array = current.split(",")
  (array(0), array(1), array(2))
)
import sqlContext.implicits._
val dataframe = splittedRDD.toDF("name", "degree", "salary")
dataframe.show()

输出是：

+-------+------+------+
|   name|degree|salary|
+-------+------+------+
|    dev|  engg| 10000|
|karthik|  engg| 20000|
+-------+------+------+

注意：(array(0), array(1), array(2)) 是一个 Scala 元组

【讨论】：

此外，spark.apache.org/docs/latest/… 的教程展示了如何手动定义架构，而不是使用不太可靠的 toDF。

spark-testing-java.readthedocs.io/en/release-1.0/Scala/…