0

それぞれ15GBのいくつかの毎日のファイルで構成される巨大なデータセットに分析を実装しようとしています。より高速にするために、テスト目的で、関連するすべてのシナリオを含む非常に小さなデータセットを作成しました。ユーザーごとに正しい一連のアクション (つまり、ログや監査に似ています) を分析する必要があります。

そのために、udf 関数を定義してから groupby を適用しました。私のユースケースを再現するコードの下：

import pyspark
from pyspark import SparkContext
import pyspark.sql.functions as F
from pyspark.sql.window import Window
from pyspark.sql.types import *
import time
sc = SparkContext.getOrCreate()
spark = pyspark.sql.SparkSession.builder.appName('example').getOrCreate()

d = spark.createDataFrame(
    [(133515, "user1", 100, 'begin'),
     (133515, "user1", 125, 'ok'),
     (133515, "user1", 150, 'ok'),
     (133515, "user1", 200, 'end'),
     (133515, "user1", 250, 'begin'),
     (133515, "user1", 300, 'end'),
     (133515, "user1", 310, 'begin'),
     (133515, "user1", 335, 'ok'),
     (133515, "user1", 360, 'ok'),
     # user1 missing END and STOPPED
     (789456, "user2", 150, 'begin'),
     (789456, "user2", 175, 'ok'),
     (789456, "user2", 200, 'end'),
     # user2 stopped
     (712346, "user3", 100, 'begin'),
     (712346, "user3", 125, 'ok'),
     (712346, "user3", 150, 'ok'),
     (712346, "user3", 200, 'end'),
     #user3 stopped
     (789456, "user4", 150, 'begin'),
     (789456, "user4", 300, 'end'),
     (789456, "user4", 350, 'begin'),
     (789456, "user4", 375, 'ok'),
     (789456, "user4", 450, 'end'),
     (789456, "user4", 475, 'ok'),
     #user4 missing BEGIN but ALIVE

    ], ("ID", "user", "epoch", "ACTION")).orderBy(F.col('epoch'))
d.show()
zip_lists = F.udf(lambda x, y: [list(z) for z in zip(x, y)], ArrayType(StringType()))

start=time.time()
d2 = d.groupBy(F.col('ID'), F.col('user'))\
.agg(zip_lists(F.collect_list('epoch'), F.collect_list('ACTION')).alias('couples'))
d2.show(50, False)
end = time.time()
print(end-start)

これにより、次の結果が得られます。

+------+-----+--------------------------------------------------------------------------------------------------------------+
|ID    |user |couples                                                                                                       |
+------+-----+--------------------------------------------------------------------------------------------------------------+
|789456|user4|[[150, begin], [300, end], [350, begin], [375, ok], [450, end], [475, ok]]                                    |
|712346|user3|[[100, begin], [125, ok], [150, ok], [200, end]]                                                              |
|133515|user1|[[100, begin], [125, ok], [150, ok], [200, end], [250, begin], [300, end], [310, begin], [335, ok], [360, ok]]|
|789456|user2|[[150, begin], [175, ok], [200, end]]                                                                         |
+------+-----+--------------------------------------------------------------------------------------------------------------+

189.9082863330841

遅すぎませんか？

condaを搭載した最新のラップトップを使用しています。conda ナビゲーターを使用して pyspark をインストールしました。

私が間違っていることはありますか？このような小さなデータセットには多すぎます

4

1 に答える 1