🔱 👩🏻‍🔬 🚩 PySpark。セッションを見つける問題の解決 🦋 🎾 🕵🏽

こんにちは、読者の皆様！数日前、AnthonyMolinaroによる本を読み直しました。レシピのコレクション」、ある章で、連続する値の範囲の開始と終了を決定することに専念するトピックに出くわしました。資料を少し読んだだけで、テスト課題のひとつとしてこの質問に出くわしたことをすぐに思い出しましたが、そのトピックは「セッションを見つける課題」と宣言されました。技術面接の秘訣は、実行された作業のレビューではなく、Sparkを使用して同様の値を取得する方法に関する面接官の質問の1つでした。インタビューの準備をしていると、会社がApache Sparkを使用している（または使用していない可能性がある）ことを知りませんでした。そのため、当時、新しいツールに関する情報を収集していませんでした。望ましい解決策はスクリプトのようなものである可能性があるという仮説を立てるだけでした。これは、Pandasライブラリを使用して作成できます。非常に離れた場所にありますが、それでも目標を達成しましたが、この組織で働くことはできませんでした。

公平を期すために、私は何年にもわたってApacheSparkの学習にほとんど進歩がなかったことに注意したいと思います。ただし、多くのアナリストがこのツールにまったく遭遇しておらず、他のアナリストも同様のインタビューを行っている可能性があるため、ベストプラクティスを読者と共有したいと思います。Sparkの専門家であれば、投稿へのコメントでいつでもより最適なコードを提案できます。

これは前文でした。このトピックの分析に直接進みましょう。まず、SQLスクリプトを作成しましょう。しかし、最初に、データベースを作成して値を入力しましょう。これはデモの例なので、SQLiteの使用をお勧めします。このデータベースは、より強力な「店内の同僚」より劣っていますが、スクリプト開発のためのその機能は私たちにとって十分です。上記の操作を自動化するために、Pythonで次のコードを記述しました。

#  
import sqlite3

#     
projects = [
    ('2020-01-01', '2020-01-02'),
    ('2020-01-02', '2020-01-03'),
    ('2020-01-03', '2020-01-04'),
    ('2020-01-04', '2020-01-05'),
    ('2020-01-06', '2020-01-07'),
    ('2020-01-16', '2020-01-17'),
    ('2020-01-17', '2020-01-18'),
    ('2020-01-18', '2020-01-19'),
    ('2020-01-19', '2020-01-20'),
    ('2020-01-21', '2020-01-22'),
    ('2020-01-26', '2020-01-27'),
    ('2020-01-27', '2020-01-28'),
    ('2020-01-28', '2020-01-29'),
    ('2020-01-29', '2020-01-30')
]

try:
    #  
    con = sqlite3.connect("projects.sqlite")
    #  
    cur = con.cursor()
    #  
    cur.execute("""CREATE TABLE IF NOT EXISTS projects (
                    proj_id INTEGER PRIMARY KEY AUTOINCREMENT,
                    proj_start TEXT,
                    proj_end TEXT)""")
    #  
    cur.executemany("INSERT INTO projects VALUES(NULL, ?,?)", projects)
    #  
    con.commit()
    #  
    cur.close()
except sqlite3.Error as err:
    print("  ", err)
finally:
    #  
    con.close()
    print("  ")

. DBeaver. , SQL .

, , , . , - . , . ().

select 
      p3.proj_group, 
      min(p3.proj_start) as date_start,
      max(p3.proj_end) as date_end,
      julianday(max(p3.proj_end))-julianday( min(p3.proj_end))+1 as delta
from
    (select 
	     p2.*,
	     sum(p2.flag)over(order by p2.proj_id) as proj_group
	from 
		(select 
		      p.proj_id , 
		      p.proj_start, 
		      p.proj_end, 
		      case 
		      when lag(p.proj_end)over(order by p.proj_id) = p.proj_start then 0 else 1 
		      end as flag
		from projects as p) as p2) as p3
group by p3.proj_group

, . . , : . , . , , lag. 0, 1. , . . , . . , ( julianday SQLite). . Spark.

, Apache Spark , Hadoop. Java, Scala R, Spark PySpark. . Google Colab, . - , . , .

Linux OpenJDK, Spark. . findspark. , .

SQLite , . , .

Spark , . , . -, , , -, . , “ Spark. ”, , , , .

, , SQL. : , ( datediff).

, . , - , , , SQL Spark. , , . .

from pyspark.sql.functions import lag
from pyspark.sql import functions as F
from pyspark.sql.window import Window
# Equivalent of Pandas.dataframe.shift() method
w = Window().partitionBy().orderBy(col("proj_id"))
df_dataframe = df.withColumn('lag', F.lag("proj_end").over(w))
#...
# Equivalent of SQL- CASE WHEN...THEN...ELSE... END
df_dataframe = df_dataframe.withColumn('flag',F.when(df_dataframe["proj_start"] == df_dataframe["lag"],0).otherwise(1))
#...
# Cumsum by column flag
w = Window().partitionBy().orderBy(col("proj_id"))
df_dataframe = df_dataframe.withColumn("proj_group", F.sum("flag").over(w))
#...
# Equivalent of SQL - GROUP BY
from pyspark.sql.functions import  min, max
df_group = df_dataframe.groupBy("proj_group").agg(min("proj_start").alias("date_start"), \
                                                  max("proj_end").alias("date_end"))
df_group = df_group.withColumn("delta", F.datediff(df_group.date_end,df_group.date_start))
df_group.show()

, . . , “” , .
これまでSparkを使用したことがない場合でも、これは空席をめぐって競争を断る理由にはなりません。PySparkの基本は、バックグラウンドがすでにPandasライブラリを使用したプログラミング経験を持っている場合、短時間で習得できます。
Sparkに関する本は不足していません。

それで全部です。すべての健康、幸運、そしてプロとしての成功！

PySpark。セッションを見つける問題の解決

More articles: