Как создать фрейм данных из определенной части текстового файла журнала, используя pyspark - PullRequest
Новая
       35

Как создать фрейм данных из определенной части текстового файла журнала, используя pyspark

0 голосов
/ 09 мая 2018

Я новичок в 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  
-------------------------------------------

Теперь я хочу разделить этот файл на несколько RDD / Dataframe, используя Spark и python (Pyspark). Я могу сделать это в Scala, используя APIHadoopFile, Теперь я хочу сделать это в 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)

Я могу создать фрейм данных, но не получаю ожидаемого вывода. Может ли кто-нибудь мне помочь.

для получения более подробной информации, смотрите прикрепленный скриншот.

Набор данных

enter image description here

1 Ответ

0 голосов
/ 13 мая 2018 
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#####
...