PySpark DataFrame 문제에서 시간 빈도를 사용하여 GroupBy

문제에 답하는 방법에는 두 가지가 있습니다. 타임 스탬프를 그룹화하려는 날짜 단위로 캐스트하거나 (설명에서 언급했듯이) 원하는 간격으로 그룹화하는 SQL 창 함수를 사용합니다.

Spark의 창 SQL 함수를 통해 월별 집계가 불가능하다는 점만 알아 두십시오.

여기에서 코드를 볼 수 있습니다. 처음 세 가지 예제는 창 SQL 함수를 사용하고 마지막 예제는 매월 타임 스탬프를 캐스팅 한 다음 모든 열을 기준으로 그룹화합니다.

df = spark.createDataFrame(
    [
        ("20-05-09 03:06:21", "PUSC_RES", "SIMPLEX", "NORTH_AL", "UE220034"),
        ("20-05-09 04:33:04", "PUSC_RES", "SIMPLEX", "SOUTH_AL", "UE220034"),
        ("20-05-09 12:04:52", "TESC_RES", "SIMPLEX", "NORTH_AL", "UE220057"),
        ("20-05-10 04:24:09", "TESC_RES", "SIMPLEX", "NORTH_AL", "UE220057"),
        ("20-05-10 04:33:04", "PUSC_RES", "SIMPLEX", "SOUTH_AL", "UE220034"),
        ("20-04-09 10:57:48", "TESC_RES", "SIMPLEX", "NORTH_AL", "UE220057"),
        ("20-04-09 12:12:26", "TESC_RES", "SIMPLEX", "NORTH_AL", "UE220057"),
        ("20-04-09 03:26:33", "PUSC_RES", "SIMPLEX", "NORTH_AL", "UE220071")
    ],
    ['CAPTUREDTIME', 'NODE', 'CHANNEL', 'LOCATION', 'TACK']
)

from pyspark.sql.functions import col, count, date_format, date_sub, date_trunc, month, next_day, to_timestamp, weekofyear, window, year

매시간

나는 여전히 이것에 대한 창 논리를 유지하므로 Spark의 모든 가능성을 참조 할 수 있습니다. 데이터 프레임을 표시하기 전에 끝 부분의 창 시작 만 선택합니다.

hourly = (
    df
    .withColumn("captured_time", to_timestamp(col('CAPTUREDTIME'), 'yy-MM-dd HH:mm:ss'))
    .groupBy(window(col("captured_time"), "1 hour").alias("captured_time"), "NODE", "CHANNEL", "LOCATION", "TACK")
    .agg(count("*"))
    .withColumn("captured_time_hour", col("captured_time.start"))
    .drop("captured_time")
)
hourly.sort("captured_time_hour").show(100, False)

매일

date_trunc기능을 통해 요일 만 고려하여 타임 스탬프를자를 수 있습니다.

daily = (
    df
    .withColumn("captured_time", to_timestamp(col('CAPTUREDTIME'), 'yy-MM-dd HH:mm:ss'))
    .withColumn("captured_time_day", date_trunc("day", col("captured_time")))
    .groupBy("captured_time_day", "NODE", "CHANNEL", "LOCATION", "TACK")
    .agg(count("*"))
)
daily.sort("captured_time_day").show(100, False)

주간

이것은 좀 더 까다 롭습니다. 먼저 next_day월요일 기능을 사용 합니다. 일요일을주의 시작으로 생각한다면 그에 따라이 코드를 업데이트하십시오. 그러나 저는 월요일을주의 시작으로 간주합니다 (내가 믿는 SQL 방언과 지역에 따라 다름).

그런 다음 weekofyear원하는대로 주 번호를 검색 하는 함수를 추가 할 수도 있습니다.

weekly = (
    df
    .withColumn("captured_time", to_timestamp(col('CAPTUREDTIME'), 'yy-MM-dd HH:mm:ss'))
    .withColumn("start_day", date_sub(next_day(col("captured_time"), "monday"), 7))
    .groupBy("start_day", "NODE", "CHANNEL", "LOCATION", "TACK")
    .agg(count("*"))
    .withColumn("start_day", to_timestamp(col("start_day")))
    .withColumn("week_of_year", weekofyear(col("start_day")))
)
weekly.sort("start_day").show(100, False)

월간 간행물

타임 스탬프를 날짜로 포맷 한 다음 타임 스탬프로 다시 캐스팅합니다. 이것은 다른 방법을 보여주기위한 것입니다. 일일 사용 사례로 타임 스탬프를자를 수 있습니다. 또한 월 이름과 약어를 추출하는 두 가지 방법을 보여줍니다. Spark 3.0.0에서 테스트되었으므로 Spark 버전을 관리하십시오.

monthly = (
    df
    .withColumn("captured_time", to_timestamp(col('CAPTUREDTIME'), 'yy-MM-dd HH:mm:ss'))
    .withColumn("captured_time_month", date_format(col('captured_time'), '1/M/yyyy'))
    .groupBy(col("captured_time_month"), "NODE", "CHANNEL", "LOCATION", "TACK")
    .agg(count("*").alias("Count TACK"))
    .withColumn("captured_time_month", to_timestamp(col("captured_time_month"), '1/M/yyyy'))
    .withColumn("month", month(col("captured_time_month")))
    .withColumn("month_abbr", date_format(col("captured_time_month"),'MMM'))
    .withColumn("full_month_name", date_format(col("captured_time_month"),'MMMM'))
)
monthly.sort("captured_time_month").show(100, False)

챠오!