SparkSQL에서 시계열 가상 데이터 테이블을 만드는 함수

소개

기능

def generate_series(day_start, day_end, sec_interval, col_name):
  # day type: str
  # day format: 'yyyy/MM/dd'
  from pyspark.sql.functions import col
  from pyspark.sql import SparkSession
  spark = SparkSession.builder.getOrCreate()

  # day_start から day_end までのタイムスタンプ df 作成
  start, stop = spark\
  .createDataFrame([(day_start, day_end)], ("start", "stop"))\
  .select([col(c).cast("timestamp").cast("long") for c in ("start", "stop")])\
  .first()

  # 指定したインターバルで select
  calendar_table = spark\
  .range(start, stop, sec_interval)\
  .select(col("id").cast("timestamp")\
  .alias(col_name))

  return calendar_table

사용 예

초당

start_date = '2020/01/01'
end_date = '2020/01/31'

# #1秒毎のテーブル
cal = generate_series(start_date, end_date, 1, 'time_stamp') 
display(cal)

1시간마다

start_date = '2020/01/01'
end_date = '2020/01/31'

# #1時間毎のテーブル
cal = generate_series(start_date, end_date, 60*60, 'time_stamp') 
display(cal)

결론

참고 링크