Как использовать partitionBy и orderBy вместе в Pyspark - PullRequest
Новая
       26

Как использовать partitionBy и orderBy вместе в Pyspark

0 голосов
/ 14 сентября 2018

Я пытаюсь разделить сеансы из набора данных lastfm, следуя ответу от https://www.arundhaj.com/blog/calculate-difference-with-previous-row-in-pyspark.html

from pyspark.sql import SparkSession
from pyspark.sql.functions import countDistinct, count, lag, to_timestamp
from pyspark.sql.window import Window


spark = SparkSession \
        .builder \
        .appName("SampleAPP") \
        .config("spark.executor.memory", "4g") \
        .getOrCreate()

fname = "inputFile.tsv"
df = spark.read.format('com.databricks.spark.csv').options(header='false', delimiter='\t').load(fname)
df = df.selectExpr("_c0 as userid", "_c1 as timestamp", "_c2 as artid", "_c3 as artname", "_c4 as traid",
                       "_c5 as traname")

df = df.withColumn('new_ts', df.timestamp.astype('Timestamp').cast("long"))
my_window = Window.partitionBy("userid").orderBy("new_ts")
df = df.withColumn("prev_value", lag(df.new_ts).over(my_window))
df.show()

, но появляется следующая ошибка: py4j.protocol.Py4JJavaError: Произошла ошибка при вызове o56.showString ,

набор данных выглядит следующим образом, 

+-----------+--------------------+--------------------+---------------+--------------------+--------------------+----------+
|     userid|           timestamp|               artid|        artname|               traid|             traname|    new_ts|
+-----------+--------------------+--------------------+---------------+--------------------+--------------------+----------+
|user_000001|2009-05-04T23:08:57Z|f1b1cf71-bd35-4e9...|      Deep Dish|                null|Fuck Me Im Famous...|1241478537|
|user_000001|2009-05-04T13:54:10Z|a7f7df4a-77d8-4f1...|           坂本龍一|                null|Composition 0919 ...|1241445250|
|user_000001|2009-05-04T13:52:04Z|a7f7df4a-77d8-4f1...|           坂本龍一|                null|Mc2 (Live_2009_4_15)|1241445124|
|user_000001|2009-05-04T13:42:52Z|a7f7df4a-77d8-4f1...|           坂本龍一|                null|Hibari (Live_2009...|1241444572|
+-----------+--------------------+--------------------+---------------+--------------------+--------------------+----------+

любая помощь в том, как решить, будет очень признательна.

Полная трассировка 

Traceback (most recent call last):
  File "code.py", line 42, in <module>
    df.show()
  File "path/to//venvs/myvenv/lib/python3.6/site-packages/pyspark/sql/dataframe.py", line 350, in show
    print(self._jdf.showString(n, 20, vertical))
  File "path/to//venvs/myvenv/lib/python3.6/site-packages/py4j/java_gateway.py", line 1257, in __call__
    answer, self.gateway_client, self.target_id, self.name)
  File "path/to//venvs/myvenv/lib/python3.6/site-packages/pyspark/sql/utils.py", line 63, in deco
    return f(*a, **kw)
  File "path/to//venvs/myvenv/lib/python3.6/site-packages/py4j/protocol.py", line 328, in get_return_value
    format(target_id, ".", name), value)
py4j.protocol.Py4JJavaError: An error occurred while calling o57.showString.
: java.lang.IllegalArgumentException
    at org.apache.xbean.asm5.ClassReader.<init>(Unknown Source)
    at org.apache.xbean.asm5.ClassReader.<init>(Unknown Source)
    at org.apache.xbean.asm5.ClassReader.<init>(Unknown Source)
    at org.apache.spark.util.ClosureCleaner$.getClassReader(ClosureCleaner.scala:46)
    at org.apache.spark.util.FieldAccessFinder$$anon$3$$anonfun$visitMethodInsn$2.apply(ClosureCleaner.scala:449)
    at org.apache.spark.util.FieldAccessFinder$$anon$3$$anonfun$visitMethodInsn$2.apply(ClosureCleaner.scala:432)
    at scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:733)
    at scala.collection.mutable.HashMap$$anon$1$$anonfun$foreach$2.apply(HashMap.scala:103)
    at scala.collection.mutable.HashMap$$anon$1$$anonfun$foreach$2.apply(HashMap.scala:103)
    at scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:230)
    at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:40)
    at scala.collection.mutable.HashMap$$anon$1.foreach(HashMap.scala:103)
    at scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:732)
    at org.apache.spark.util.FieldAccessFinder$$anon$3.visitMethodInsn(ClosureCleaner.scala:432)
    at org.apache.xbean.asm5.ClassReader.a(Unknown Source)
    at org.apache.xbean.asm5.ClassReader.b(Unknown Source)
    at org.apache.xbean.asm5.ClassReader.accept(Unknown Source)
    at org.apache.xbean.asm5.ClassReader.accept(Unknown Source)
    at org.apache.spark.util.ClosureCleaner$$anonfun$org$apache$spark$util$ClosureCleaner$$clean$14.apply(ClosureCleaner.scala:262)
    at org.apache.spark.util.ClosureCleaner$$anonfun$org$apache$spark$util$ClosureCleaner$$clean$14.apply(ClosureCleaner.scala:261)
    at scala.collection.immutable.List.foreach(List.scala:381)
    at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:261)
    at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:159)
    at org.apache.spark.SparkContext.clean(SparkContext.scala:2299)
    at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1.apply(RDD.scala:797)
    at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1.apply(RDD.scala:796)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
    at org.apache.spark.rdd.RDD.withScope(RDD.scala:363)
    at org.apache.spark.rdd.RDD.mapPartitions(RDD.scala:796)
    at org.apache.spark.sql.execution.window.WindowExec.doExecute(WindowExec.scala:302)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:131)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:127)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:155)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
    at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:152)
    at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:127)
    at org.apache.spark.sql.execution.InputAdapter.inputRDDs(WholeStageCodegenExec.scala:371)
    at org.apache.spark.sql.execution.ProjectExec.inputRDDs(basicPhysicalOperators.scala:41)
    at org.apache.spark.sql.execution.BaseLimitExec$class.inputRDDs(limit.scala:62)
    at org.apache.spark.sql.execution.LocalLimitExec.inputRDDs(limit.scala:97)
    at org.apache.spark.sql.execution.WholeStageCodegenExec.doExecute(WholeStageCodegenExec.scala:605)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:131)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:127)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:155)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
    at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:152)
    at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:127)
    at org.apache.spark.sql.execution.SparkPlan.getByteArrayRdd(SparkPlan.scala:247)
    at org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:337)
    at org.apache.spark.sql.execution.CollectLimitExec.executeCollect(limit.scala:38)
    at org.apache.spark.sql.Dataset.org$apache$spark$sql$Dataset$$collectFromPlan(Dataset.scala:3273)
    at org.apache.spark.sql.Dataset$$anonfun$head$1.apply(Dataset.scala:2484)
    at org.apache.spark.sql.Dataset$$anonfun$head$1.apply(Dataset.scala:2484)
    at org.apache.spark.sql.Dataset$$anonfun$52.apply(Dataset.scala:3254)
    at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:77)
    at org.apache.spark.sql.Dataset.withAction(Dataset.scala:3253)
    at org.apache.spark.sql.Dataset.head(Dataset.scala:2484)
    at org.apache.spark.sql.Dataset.take(Dataset.scala:2698)
    at org.apache.spark.sql.Dataset.showString(Dataset.scala:254)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:564)
    at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
    at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
    at py4j.Gateway.invoke(Gateway.java:282)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:238)
    at java.base/java.lang.Thread.run(Thread.java:844)


Process finished with exit code 1

датафрейм работает после следующей строки, 

df = df.withColumn('new_ts', df.timestamp.astype('Timestamp').cast("long"))

enter image description here

РЕШЕНИЕ: Я пытался исправить это в своей локальной среде, но, к сожалению, я не смог.

использовал образ докера из https://github.com/MinerKasch/training-docker-pyspark и выполнен в Jupyter Notebook, и тот же код работает.

Большое спасибо за помощь.

Ответы [ 2 ]

0 голосов
/ 14 сентября 2018

Трудно воспроизвести вашу ошибку.Когда я запускаю следующий код, который в значительной степени соответствует вашей логике, я не получаю никаких ошибок: 

test_df = spark.createDataFrame([
    (1,datetime.date(2017, 4, 1)),(2,datetime.date(2017, 4, 2)),(3,datetime.date(2017, 4, 3)),(1,datetime.date(2017, 4, 2)),(1,datetime.date(2017, 4, 4)),                   
    ], ("_c0","_c1"))


test_df = test_df.selectExpr("_c0 as userid", "_c1 as timestamp")
test_df = test_df.withColumn('new_ts', test_df.timestamp.astype('Timestamp').cast("long"))
my_window = Window.partitionBy("userid").orderBy("new_ts")
test_df = test_df.withColumn("prev_value", lag(test_df.new_ts).over(my_window))
test_df.show()

Вывод: 

+------+----------+----------+----------+
|userid| timestamp|    new_ts|prev_value|
+------+----------+----------+----------+
|     1|2017-04-01|1491004800|      null|
|     1|2017-04-02|1491091200|1491004800|
|     1|2017-04-04|1491264000|1491091200|
|     3|2017-04-03|1491177600|      null|
|     2|2017-04-02|1491091200|      null|
+------+----------+----------+----------+

Так что я думаю, что ваша проблемав ваших данных, и у вас есть специальный символ в одном из файлов.Я также воссоздал небольшой tsv-файл и полностью использовал ваш код, который работал для меня (см. Версию 2).

Версия 2: 

df = spark.read.format('com.databricks.spark.csv').options(header='false', delimiter='\t').load("/home/spark/test.csv")
df = df.selectExpr("_c0 as userid", "_c1 as timestamp")

df = df.withColumn('new_ts', df.timestamp.astype('Timestamp').cast("long"))
my_window = Window.partitionBy("userid").orderBy("new_ts")
df = df.withColumn("prev_value", lag(df.new_ts).over(my_window))
df.show()

Вывод 2 

+--------+--------------------+----------+----------+
|  userid|           timestamp|    new_ts|prev_value|
+--------+--------------------+----------+----------+
|user_001|2009-05-04T13:52:04Z|1241445124|      null|
|user_001|2009-05-04T13:54:10Z|1241445250|1241445124|
|user_001|2009-05-04T23:08:57Z|1241478537|1241445250|
+--------+--------------------+----------+----------+
0 голосов
/ 14 сентября 2018

Ошибка не в синтаксисе разбиения окна.Поскольку spark выполняет ленивую оценку, вы получаете ошибку в show ().Значением ошибки может быть любое преобразование, которое было сделано до этого момента.

Но я полагаю, что ошибка в основном вызвана java.lang.OutOfMemoryError: Пространство кучи Java , поскольку разбиение - это тяжелая операция с памятью.Я предлагаю прочитать полную трассировку ошибок.Если мой прогноз верен, вам следует увеличить память, чтобы избежать этой проблемы.

...