pyspark案例系列4-dataframe输出到单个文件夹的解决方案
Posted 只是甲
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了pyspark案例系列4-dataframe输出到单个文件夹的解决方案相关的知识,希望对你有一定的参考价值。
文章目录
一. 需求
今天有个需求,需要将pyspark里面的dataframe输入到单个txt文件中。
我们知道pyspark是分布式的,所以写文件的时候也是多线程操作,此时就会生成多个不同的文件。
二. 解决方案
2.1 dataframe写本地磁盘
查阅资料发现保存到本地文件系统(file:///)只有再local模式下才能生效,在cluster模式下(不论是yarn-client还是yarn-cluster)都无法使用
2.1.1 csv文件格式
/home/spark/dir1 这个目录由程序来创建,如果已存在会报错
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from pyspark.sql import SparkSession
spark = SparkSession \\
.builder \\
.appName("Python Spark SQL basic example") \\
.master("local") \\
.config("spark.some.config.option", "some-value") \\
.getOrCreate()
df1 = spark.sql("select * from test.emp")
df1.coalesce(1).write.format("csv").options(header='true', inferschema='true').save("file:///home/spark/dir1")
运行程序:
spark-submit --master local 1.py
测试记录:
2.1.1 text文件格式
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from pyspark.sql import SparkSession
spark = SparkSession \\
.builder \\
.appName("Python Spark SQL basic example") \\
.master("local") \\
.config("spark.some.config.option", "some-value") \\
.getOrCreate()
df1 = spark.sql("select * from test.emp")
df1.coalesce(1).write.format("text").options(header='true', inferschema='true').save("file:///home/spark/dir2")
运行报错:
pyspark.sql.utils.AnalysisException: u’Text data source does not support int data type.;’
需要调整代码,写text之前将dataframe保存为字符串类型
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from pyspark.sql import SparkSession
from pyspark.sql.functions import concat_ws
spark = SparkSession \\
.builder \\
.appName("Python Spark SQL basic example") \\
.master("local") \\
.config("spark.some.config.option", "some-value") \\
.getOrCreate()
df1 = spark.sql("select * from test.emp")
df2 = df1.select(concat_ws(',',*df1.columns).alias('data'))
df2.coalesce(1).write.format("text").options(header='true', inferschema='true').save("file:///home/spark/dir2")
2.1 dataframe写HDFS
以写csv格式为例
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from pyspark.sql import SparkSession
spark = SparkSession \\
.builder \\
.appName("Python Spark SQL basic example") \\
.master("local") \\
.config("spark.some.config.option", "some-value") \\
.getOrCreate()
df1 = spark.sql("select * from test.emp")
df1.coalesce(1).write.format("csv").options(header='true', inferschema='true').save("hdfs://hp1:8020/user/juzhen/dir1")
测试记录:
参考:
- https://blog.csdn.net/jingyi130705008/article/details/108236217
以上是关于pyspark案例系列4-dataframe输出到单个文件夹的解决方案的主要内容,如果未能解决你的问题,请参考以下文章
Python应用实战案例-pyspark库从安装到实战保姆级讲解
PySpark - ALS 输出中的 RDD 到 DataFrame
PySpark 结构化流式处理:将查询的输出传递到 API 端点