代码之家 › 专栏 › 技术社区 › YQ.Wang

使用pyspark会话从本地文件读取时,如何跳过一些行?

pyspark apache-spark python

2

YQ.Wang · 技术社区 · 7 年前

我正在使用pyspark从本地读取和处理一些数据 .plt 文件夹。文件如下:

Geolife trajectory
WGS 84
Altitude is in Feet
Reserved 3
0,2,255,My Track,0,0,2,8421376
0
39.984094,116.319236,0,492,39744.2451967593,2008-10-23,05:53:05
39.984198,116.319322,0,492,39744.2452083333,2008-10-23,05:53:06
39.984224,116.319402,0,492,39744.2452662037,2008-10-23,05:53:11
39.984211,116.319389,0,492,39744.2453240741,2008-10-23,05:53:16
......

如上图所示,我对开始的6行不感兴趣,我想要的是从第7行开始的行。所以我想用spark会话从第7行读取这个文件。以下是我尝试但失败的代码:

from pyspark.sql import SparkSession
session = SparkSession.builder.appName('file reader').master('local[*]').getOrCreate()
df = session.read.\
     option('delimiter', ',').\
     option('header', 'false').\
     csv('test.plt')
df.show()

有人能给我一些建议吗?谢谢你的关注。

3 回复 | 直到 7 年前

1

3

Arnon Rotem-Gal-Oz 7 年前

from pyspark.sql.types import *
from pyspark.sql import SparkSession
session = SparkSession.builder.appName('file reader').master('local[*]').getOrCreate()
schema = StructType([StructField("a", FloatType()),
                     StructField("b", FloatType()),
                     StructField("c", IntegerType()),
                     StructField("d", IntegerType()),
                     StructField("e", FloatType()),
                     StructField("f", StringType()),
                     StructField("g", StringType())])
df=session.read.option('mode','DROPMALFORMED').csv('test.plt',schema)

2

1

cph_sto 7 年前

此外,还有@ Arnon Rotem-Gal-Oz 我们也可以利用任何列的某些特殊属性(如果存在的话)。

在 YQ. Wang's 数据,我们可以看到 6th 列是日期,而 第六 列中 header 也将是 date . 所以,我们的想法是检查 第六 列。 to_date() 转换A string 到 日期 . 如果此列不是 日期 然后 ToDATA() 将返回 Null 我们将使用 .where() 条款

from pyspark.sql.functions import to_date
from pyspark.sql.types import FloatType, StringType, StructType, StructField
df = spark.read.schema(schema)\
                    .format("csv")\
                    .option("header","false")\
                    .option("sep",',')\
                    .load('test.plt')\
                    .where(to_date(col('f'),'yyyy-MM-dd').isNotNull())
df.show()
+---------+----------+----+---+---------+----------+--------+
|        a|         b|   c|  d|        e|         f|       g|
+---------+----------+----+---+---------+----------+--------+
|39.984093| 116.31924|   0|492|39744.246|2008-10-23|05:53:05|
|  39.9842| 116.31932|   0|492|39744.246|2008-10-23|05:53:06|
|39.984222|116.319405|   0|492|39744.246|2008-10-23|05:53:11|
| 39.98421| 116.31939|   0|492|39744.246|2008-10-23|05:53:16|
+---------+----------+----+---+---------+----------+--------+

这种方法也有缺点,比如 日期 丢失了,然后整个行被过滤掉。

3

0

Dave Canton 7 年前

假设从第7行开始的数据符合您所示的模式:

from pyspark.sql import SparkSession
session = SparkSession.builder.appName('file reader').master('local[*]').getOrCreate()
data = session.read.textFile('test.plt')

header = data.head(6)  # the first six rows

filtered = data.filter(row => row != header)
               .withColumn("a", split(col("value"), ",").getItem(0))
               .withColumn("b", split(col("value"), ",").getItem(1))
               .withColumn("c", split(col("value"), ",").getItem(2))
               .withColumn("d", split(col("value"), ",").getItem(3))
               .withColumn("e", split(col("value"), ",").getItem(4))
               .withColumn("f", split(col("value"), ",").getItem(5))
               .withColumn("g", split(col("value"), ",").getItem(6))
               .drop("value")