如何在Java中打印整個數(shù)據(jù)幀而不會耗盡內存？Dataset<Row> df = ...我知道那件事：df.show() 將顯示數(shù)據(jù)幀，但是對于足夠大的數(shù)據(jù)幀，這可能會耗盡內存。我知道我可以使用以下內容限制內容：df.show(rowCount, false)但是想要打印整個數(shù)據(jù)幀，我不想限制內容...我試過：df.foreachPartition(iter -> { while(iter.hasNext()){ System.out.println(rowIter.next().mkString(",");) }});但這將打印在每個相應的節(jié)點上，而不是在驅動程序上...如果有什么方法可以打印驅動程序中的所有內容而不會耗盡內存？
2 回答
小怪獸愛吃肉
TA貢獻1852條經驗 獲得超1個贊
AFAIK，打印數(shù)據(jù)框的想法是查看數(shù)據(jù)。
不建議根據(jù)內存不足的數(shù)據(jù)幀大小打印大型數(shù)據(jù)幀。
我會提供以下方法，如果你想看到內容，那么你可以保存在hive表中并查詢內容?；驅懭肟勺x的csv或json
例子：
1） 保存在蜂巢表中
df.write.mode("overwrite").saveAsTable("database.tableName")稍后從配置單元表查詢。
2） csv 或 json
df.write.csv("/your/location/data.csv")
df.write.json("/your/location/data.json")如果您希望使用單個文件，上述內容將生成多個零件文件（但這會再次將數(shù)據(jù)移動到一個節(jié)點，除非您絕對需要它，否則不鼓勵這樣做）coalesce(1)
另一種選擇是使用localIterator逐行打印，請參閱此處，這也將數(shù)據(jù)傳輸?shù)焦?jié)點...因此它不是一個好主意
楊魅力
TA貢獻1811條經驗 獲得超6個贊
您將不得不將所有數(shù)據(jù)帶到驅動程序，這將:(一點地占用您的內存...
解決方案可能是拆分數(shù)據(jù)幀并在驅動程序中逐個打印。當然，這取決于數(shù)據(jù)本身的結構，它看起來像這樣：
long count = df.count();
long inc = count / 10;
for (long i = 0; i < count; i += inc) {
Dataset<Row> filteredDf =
df.where("id>=" + i + " AND id<" + (i + inc));
List<Row> rows = filteredDf.collectAsList();
for (Row r : rows) {
System.out.printf("%d: %s\n", r.getAs(0), r.getString(1));
}
}
我將數(shù)據(jù)集拆分為10，但我知道我的id從1到100...
完整的示例可以是：
package net.jgp.books.sparkWithJava.ch20.lab900_splitting_dataframe;
import java.util.ArrayList;
import java.util.List;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
/**
* Splitting a dataframe to bring it back to the driver for local
* processing.
*
* @author jgp
*/
public class SplittingDataframeApp {
/**
* main() is your entry point to the application.
*
* @param args
*/
public static void main(String[] args) {
SplittingDataframeApp app = new SplittingDataframeApp();
app.start();
}
/**
* The processing code.
*/
private void start() {
// Creates a session on a local master
SparkSession spark = SparkSession.builder()
.appName("Splitting a dataframe to collect it")
.master("local")
.getOrCreate();
Dataset<Row> df = createRandomDataframe(spark);
df = df.cache();
df.show();
long count = df.count();
long inc = count / 10;
for (long i = 0; i < count; i += inc) {
Dataset<Row> filteredDf =
df.where("id>=" + i + " AND id<" + (i + inc));
List<Row> rows = filteredDf.collectAsList();
for (Row r : rows) {
System.out.printf("%d: %s\n", r.getAs(0), r.getString(1));
}
}
}
private static Dataset<Row> createRandomDataframe(SparkSession spark) {
StructType schema = DataTypes.createStructType(new StructField[] {
DataTypes.createStructField(
"id",
DataTypes.IntegerType,
false),
DataTypes.createStructField(
"value",
DataTypes.StringType,
false) });
List<Row> rows = new ArrayList<Row>();
for (int i = 0; i < 100; i++) {
rows.add(RowFactory.create(i, "Row #" + i));
}
Dataset<Row> df = spark.createDataFrame(rows, schema);
return df;
}
}
你認為這可以幫助嗎？
它不像將其保存在數(shù)據(jù)庫中那樣優(yōu)雅，但它可以避免在體系結構中使用其他組件。這段代碼不是很通用，我不確定你能在當前版本的Spark中使它成為通用的。
添加回答
舉報
0/150
提交
取消