我正在从S3中以parquet格式读取数据,然后将这些数据作为DataFrame处理。
问题是如何有效地遍历DataFrame中的行?我知道方法collect将数据加载到内存中,因此,尽管我的DataFrame并不大,但我还是希望避免将完整的数据集加载到内存中。如何优化给定的代码?
另外,我正在使用索引来访问DataFrame中的列。我可以按列名访问它们吗(我知道它们)?

DataFrame parquetFile = sqlContext.read().parquet("s3n://"+this.aws_bucket+"/"+this.aws_key_members);
parquetFile.registerTempTable("mydata");
DataFrame eventsRaw = sqlContext.sql("SELECT * FROM mydata");
Row[] rddRows = eventsRaw.collect();
for (int rowIdx = 0; rowIdx < rddRows.length; ++rowIdx)
{
   Map<String, String> props = new HashMap<>();
   props.put("field1", rddRows[rowIdx].get(0).toString());
   props.put("field2", rddRows[rowIdx].get(1).toString());
   // further processing
}

最佳答案

您可以在spark中使用Map功能。
您可以迭代整个数据框架而无需收集数据集/数据框架。

Dataset<Row> namesDF = spark.sql("SELECT name FROM parquetFile WHERE age
BETWEEN 13 AND 19");
Dataset<String> namesDS = namesDF.map((MapFunction<Row, String>) row -> "Name:" + row.getString(0),Encoders.STRING());

namesDS.show();


如果您要执行的操作很复杂,则可以创建一个map函数。

 // Map function
 Row doSomething(Row row){
   // get column
     String field = row.getAs(COLUMN)
// construct a new row and add all the existing/modified columns in the row .
return row.
    }


现在可以将此映射函数调用到数据框的映射函数中

StructType structType = dataset.schema();
namesDF.map((MapFunction<Row, Row>)dosomething,
        RowEncoder.apply(structType))


资料来源:https://spark.apache.org/docs/latest/sql-data-sources-parquet.html

09-10 07:53
查看更多