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

소개

제목대로 SparkSQL에서 시계열 가상 데이터 테이블을 만드는 함수입니다. 다음과 같은 분을 상정.

최근 Spark 만지기 시작했다

시계열 데이터로 Window 처리를 일괄 적으로 행하고 싶다

SparkSQL에서 보통 쿼리를 쓸 수 있다고 들었지만,

MySQL에서의 일반적인 메소드는 사용할 수 없을 것 같습니다

PostgreSQL의 Generate Series를 사용할 수 있으면 좋겠는데

기능

이런 식으로 썼습니다. 시작일, 종료일, 초 단위의 간격, 테이블의 행명을 인수로 하고 있습니다.

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)

결론

우선은 작동하지만,

이 함수 사용하면 일발이야.

이런 비효율적인 일로…

라는 분, 정보 주시면 매우 기쁩니다.

참고 링크

Spark Programming Guide