pyspark有条件地解析固定宽度的文本文件
我有一个固定宽度的文件，我不知道它的格式，直到它中的某个变量检查某个变量是'01'还是'02'。所以我想创建这样的东西：
myreport= spark.read.text("/mnt/path/mydata")
myreport= myreport.select(myreport.value.substr(1,3).alias('client'),
myreport.value.substr(4,2).alias('rptnum'),
if rptnum = '01', then
myreport.value.substr(6,2).cast('integer').alias('mo1'),
myreport.value.substr(8,2).cast('integer').alias('mo2'),
myreport.value.substr(12,2).cast('integer').alias('mo3'),
Else
myreport.value.substr(6,2).cast('integer').alias('mo1'),
myreport.value.substr(8,2).cast('integer').alias('mo2'),
myreport.value.substr(12,2).cast('integer').alias('mo3'),
myreport.value.substr(14,2).cast('integer').alias('mo4'),
myreport.value.substr(16,2).cast('integer').alias('mo5'),
myreport.value.substr(18,2).cast('integer').alias('mo6'),
如果rpt编号不是01，基本上列的数量会翻倍。不确定如何在pyspark中执行此操作
写回答
-
社区小助手是spark中国社区的管理员，我会定期更新直播回顾等资料和文章干货，还整合了大家在钉群提出的有关spark的问题及回答。编写从df.rdd.map()中调用的函数并转换/解析每一行。您可以创建相同数量的列，但在一种情况下，某些列将为null。filter()在rptnum上使用，您可以分离行并选择相应的列。
from pyspark.sql.functions import *
from pyspark.sql import *def transformRow(row):
value = row['value'] client = value[1:4] rptnum = value[4:6] rowDict = {'client': client, 'rptnum': rptnum,'mo1': None,'mo2': None,'mo3': None,'mo4': None,'mo5': None,'mo6': None} rowDict['mo1'] = value[6:8] rowDict['mo2'] = value[8:10] rowDict['mo3'] = value[10:12] if rptnum != '01' : rowDict['mo4'] = value[12:14] rowDict['mo5'] = value[14:16] rowDict['mo6'] = value[16:18] return Row(**rowDict)myreport= spark.read.text("/mnt/path/mydata")
myreport = myreport.rdd.map(transformRow).toDF()rpt1 = myreport.filter(col("rptnum") == '01').select("mo1","mo2","mo3")
rpt2 = myreport.filter(col("rptnum") != '01')2019-07-17 23:23:00赞同 展开评论 打赏
版权声明：本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
