python - 如何使用 pyspark 从文本日志文件的特定部分创建数据框
问题描述
我是 pyspark 的新手......我有一个很大的日志文件,其中包含如下数据:
sfdfd
fsdfsdffdhfgjgfjkyklhljk,erygrt,tegtyryu,.
sgsgggggfsdf
==========================================
Roll Name class
==========================================
1 avb wer21g2
------------------------------------------
===========================================
empcode Emnname Dept Address
===========================================
12d sf sdf22 dghsjf
asf2 asdfw2 df21df fsfsfg
dsf21 sdf2 df2 sdgfsgf
-------------------------------------------
现在我想使用 Spark 和 python (Pyspark) 将此文件拆分为多个 RDD/Dataframe。我可以使用 APIHadoopFile 在 Scala 中执行此操作,现在我想在 Pyspark 中执行此操作。谁可以帮我这个事。
加速输出为:
Roll Name clas
1 avb wer21g2
empcode Emnname Dept Address
12d sf sdf22 dghsjf
asf2 asdfw2 df21df fsfsfg
dsf21 sdf2 df2 sdgfsgf
这是我尝试过的代码:
with open(path) as f:
out = []
for line in f:
if line.rstrip() == findStr:
tmp = []
tmp.append(line)
for line in f:
# print(line)
if line.rstrip() == EndStr:
out.append(tmp)
break
tmp.append(line)
f.close()
SMN_df = spark.createDataFrame(tmp, StringType()).show(truncate=False)
我能够创建数据框但没有得到预期的输出。任何人都可以帮助我。
有关更多详细信息,请参阅随附的屏幕截图。
数据集
解决方案
from pyspark.sql import SparkSession
import re
spark=SparkSession.Builder.config("spark.sql.warehouse.dir","file://C:/temp")
.appName("SparkSQL").getOrCreate()
path="C:/Users/Rudrashis/Desktop/test2.txt"
Txtpath="L:/SparkScala/test.csv"
EndStr="---------------------------------"
FilterStr="================================="
def prepareDataset(Findstr):
with open(path) as f:
out=[]
for line in f:
if line.rstrip()==Findstr:
tmp=[]
tmp.append(re.sub("\s+",",",line.strip()))
for line in f:
if line.rstrip()==EndStr:
out.append(tmp)
break
tmp.append(re.sub("\s+",",",line.strip()))
return (tmp)
f.close()
def Makesv(Lstcommon):
with open("test.csv","w")as outfile:
for entries in map(str.strip(),Lstcommon):
outfile.write(entries)
outfile.close()
###For 1st block################
LstStudent=[]
LstStudent=prepareDataset("Roll Name Class")
LstStudent.list(filter(lambda a: a!=FilterStr,LstStudent))
createStudent=Makesv(LstStudent)
Student_DF=spark.read.format('com.databricks.spark.csv')
.options(header="true",inferschema="true").load(Txtpath)
Student_DF.show(truncate=False)
######### end 1st block####
#####2nd block start####
LstEmp=[]
LstEmp=prepareDataset("empcode Emnname Dept Address")
LstEmp.list(filter(lambda a: a!=FilterStr,LstEmp))
CreateEmp=Makesv(LstEmp)
Emp_DF=spark.read.format('com.databricks.spark.csv')
.options(header="true",inferschema="true").load(Txtpath)
Emp_DF.show(truncate=False)
##### end of 2nd block#####
推荐阅读
- c++ - 如何将 C++ 头文件中的定义编译为 LLVM JIT 模块中的 IR?
- php - 未捕获的错误:找不到类“SimpleXMLElement”
- oracle - 在 Apache Nifi 中引用另一个查询的结果
- python - Imread openCV无法在windows中读取图像
- react-native - 确保您正在运行 Metro 服务器或您的捆绑包“index.android.bundle”已正确打包以供发布
- ruby - 在Ruby中,是否可以在不同的模块中同时拥有父类和子类
- mysql - 在 MariaDB 上恢复我的 root 用户的 DBA 权限
- python - 在 PyGlet 中制作窗口
- http - 未处理的异常:在收到完整标头之前连接已关闭
- java - 为什么 CodeChef 只接受第一个而不是第二个?codechef 中的问题代码 - FCTRL2 竞赛代码 - 实践