基于登录和注销时间戳的FInding活动用户使用PySpark

Question

有一个包含物联网数据的数据框架。列是设备id(唯一id)，连接在(时间戳)，断开连接在(时间戳)。我必须让活跃的用户根据时间。首先，我们需要从连接日期和断开日期获取日期，然后创建日期列。然后我们必须创建另一个列，比如"active users“，它必须显示当天活动设备的数量。

示例：

date          active users
20-08-2021     1 (it will check connection and disconnection status in dataframe)

Answer 1

我已经为您创建了示例数据和脚本。示例数据和您的尝试是您的工作。通常，在堆栈溢出上，人们不会做这样的事情，这样的问题可能会因为缺少详细信息或需要调试的脚本而被关闭。就因为我有心情..。

沿着这些路线做点什么..。

示例数据：

from pyspark.sql import functions as F
df = spark.createDataFrame(
    [(101, '2020-01-01 13:01:01', '2020-01-01 14:01:01'),
     (101, '2020-01-01 15:01:01', '2020-01-01 16:01:01'),  # same user 2nd time
     (101, '2020-01-01 17:01:01', '2020-01-02 11:01:01'),  # same user 3rd time, but overnight
     (102, '2020-01-02 05:01:01', '2020-01-02 07:01:01'),
     (103, '2020-01-02 05:01:01', '2020-01-02 07:01:01'),
     (104,                  None, '2020-01-02 07:01:01'),  # No connected_at time
     (105, '2020-01-04 10:01:01',                  None)], # No disconnected_at time
    ['device_id', 'connected_at', 'disconnected_at'])

剧本：

connection_timeout_days = -1 # If negative, not applied
logged_days_if_no_start_date = 3


# Change timestamps to dates
df = df.withColumn('connected_at', F.to_date('connected_at')) \
       .withColumn('disconnected_at', F.to_date('disconnected_at'))

# If no connected_at, subtract default number2
df = df.withColumn('connected_at', F.coalesce('connected_at',
                                   F.date_sub('disconnected_at', logged_days_if_no_start_date)))
# If no disconnected_at, add today's date or timeout days
if connection_timeout_days < 0:
    df = df.withColumn('disconnected_at', F.coalesce('disconnected_at', F.current_date()))
else:
    df = df.withColumn(
        'disconnected_at',
        F.coalesce('disconnected_at',
                   F.least(F.current_date(), F.date_add('connected_at', connection_timeout_days))))

# Create a separate df with dates for the whole period
min_date = df.agg(F.min('connected_at')).head()[0]
max_date = df.agg(F.max('disconnected_at')).head()[0]
df_dates = spark.range(1).select(
    F.explode(F.sequence(F.lit(min_date), F.lit(max_date))).alias('date')
)

# Transform original df - count distinct users per dates
df = (df
    .select('device_id',
            F.explode(F.sequence('connected_at', 'disconnected_at')).alias('date'))
    .groupBy('date')
    .agg(F.countDistinct('device_id').alias('device_distinct_count'))
)

# Join both dfs
df = df_dates.join(df, 'date', 'left') \
             .fillna(0).sort('date')

结果：

df.show()
# +----------+---------------------+
# |      date|device_distinct_count|
# +----------+---------------------+
# |2019-12-30|                    1|
# |2019-12-31|                    1|
# |2020-01-01|                    2|
# |2020-01-02|                    4|
# |2020-01-03|                    0|
# |2020-01-04|                    1|
# |2020-01-05|                    1|
# |2020-01-06|                    1|
# |2020-01-07|                    1|
# |2020-01-08|                    1|
# |2020-01-09|                    1|
# |2020-01-10|                    1|
# |2020-01-11|                    1|
# |2020-01-12|                    1|
# |2020-01-13|                    1|
# |2020-01-14|                    1|
# |2020-01-15|                    1|
# |2020-01-16|                    1|
# |2020-01-17|                    1|
# |2020-01-18|                    1|
# +----------+---------------------+
# only showing top 20 rows