我是Apache Beam的新手，并实现了我的第一个管道。
但现在我到了一个地步，我很困惑如何将开窗和加入结合起来。
问题定义：
我有两个数据流，一个是用户的页面浏览量，另一个则是用户的请求。它们共享描述用户会话的密钥session_id，但每个会话都有其他附加数据。
目标是在请求发生之前计算会话中的页面浏览量。这意味着，我想要一个数据流，其中包含每个请求以及请求之前的页面浏览次数。只要有最后5分钟的浏览量就足够了。
我尝试了什么
为了加载请求，我使用了这个片段，它从pubsub订阅加载请求，然后提取session_id作为密钥。最后，我应用了一个窗口，它在收到每个请求时直接发出请求。
requests = (p
| 'Read Requests' >> (
beam.io.ReadFromPubSub(subscription=request_sub)
| 'Extract' >> beam.Map(lambda x: json.loads(x))
| 'Session as Key' >> beam.Map(lambda request: (request['session_id'], request))
| 'Window' >> beam.WindowInto(window.SlidingWindows(5 * 60, 1 * 60, 0),
trigger=trigger.AfterCount(1),
accumulation_mode=trigger.AccumulationMode.DISCARDING
)
)
)
类似地，这个片段加载页面视图，它应用了一个滑动窗口，每当页面视图进入时，就会累积发出该窗口。
pageviews = (p
| 'Read Pageviews' >> (
beam.io.ReadFromPubSub(subscription=pageview_sub)
| 'Extract' >> beam.Map(lambda x: json.loads(x))
| 'Session as Key' >> beam.Map(lambda pageview: (pageview['session_id'], pageview))
| 'Window' >> beam.WindowInto(
windowfn=window.SlidingWindows(5 * 60, 1 * 60, 0),
trigger=trigger.AfterCount(1),
accumulation_mode=trigger.AccumulationMode.ACCUMULATING
)
)
)
为了申请加入，我尝试了
combined = (
{
'requests': requests,
'pageviews': pageviews
}
| 'Merge' >> beam.CoGroupByKey()
| 'Print' >> beam.Map(print)
)
当我运行这个管道时，在合并的行中从来没有包含请求和页面视图的行，只有其中一行。
我的想法是在请求之前过滤掉页面浏览量，然后在分组之后计数。我需要做什么？我想我的问题在于窗口和触发策略。
同样重要的是，以低延迟处理请求，可能会丢弃迟到的页面视图。
我自己找到了一个解决方案，下面是它，以防有人感兴趣：
想法
诀窍是使用beam.Flatten操作组合两个流，并使用Stateful DoFn来计算一个请求之前的页面浏览量。每个流都包含json字典。我通过使用{'request' : request}和{'pageview' : pageview}作为周围块来嵌入它们，这样我就可以在Stateful DoFn中将不同的事件分开。我还计算了第一次页面浏览的时间戳和自第一次页面查看以来的秒数。流必须使用session_id作为密钥，以便Stateful DoFn仅接收一个会话的所有事件。
代码
首先，这是管道代码：
# Beam pipeline, that extends requests by number of pageviews before request in that session
with beam.Pipeline(options=options) as p:
# The stream of requests
requests = (
'Read from PubSub subscription' >> beam.io.ReadFromPubSub(subscription=request_sub)
| 'Extract JSON' >> beam.ParDo(ExtractJSON())
| 'Add Timestamp' >> beam.ParDo(AssignTimestampFn())
| 'Use Session ID as stream key' >> beam.Map(lambda request: (request['session_id'], request))
| 'Add type of event' >> beam.Map(lambda r: (r[0], ('request', r[1])))
)
# The stream of pageviews
pageviews = (
'Read from PubSub subscription' >> beam.io.ReadFromPubSub(subscription=pageview_sub)
| 'Extract JSON' >> beam.ParDo(ExtractJSON())
| 'Add Timestamp' >> beam.ParDo(AssignTimestampFn())
| 'Use Session ID as stream key' >> beam.Map(lambda pageview: (pageview['session_id'], pageview))
| 'Add type of event' >> beam.Map(lambda p: (p[0], ('pageview', p[1])))
)
# Combine the streams and apply Stateful DoFn
combined = (
(
p | ('Prepare requests stream' >> requests),
p | ('Prepare pageviews stream' >> pageviews)
)
| 'Combine event streams' >> beam.Flatten()
| 'Global Window' >> beam.WindowInto(windowfn=window.GlobalWindows(),
trigger=trigger.AfterCount(1),
accumulation_mode=trigger.AccumulationMode.DISCARDING)
| 'Stateful DoFn' >> beam.ParDo(CountPageviews())
| 'Compute processing delay' >> beam.ParDo(LogTimeDelay())
| 'Format for BigQuery output' >> beam.ParDo(FormatForOutputDoFn())
)
# Write to BigQuery.
combined | 'Write' >> beam.io.WriteToBigQuery(
requests_extended_table,
schema=REQUESTS_EXTENDED_TABLE_SCHEMA,
create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND)
有趣的部分是使用beam.Flatten和应用有状态DoFnCountPageviews()的两个流的组合
以下是使用的自定义DoFns:的代码
# This DoFn just loads a json message
class ExtractJSON(beam.DoFn):
def process(self, element):
import json
yield json.loads(element)
# This DoFn adds the event timestamp of messages into it json elements for further processing
class AssignTimestampFn(beam.DoFn):
def process(self, element, timestamp=beam.DoFn.TimestampParam):
import datetime
timestamped_element = element
timestamp_utc = datetime.datetime.utcfromtimestamp(float(timestamp))
timestamp = timestamp_utc.strftime("%Y-%m-%d %H:%M:%S")
timestamped_element['timestamp_utc'] = timestamp_utc
timestamped_element['timestamp'] = timestamp
yield timestamped_element
# This class is a stateful dofn
# Input elements should be of form (session_id, {'event_type' : event}
# Where events can be requests or pageviews
# It computes on a per session basis the number of pageviews and the first pageview timestamp
# in its internal state
# Whenever a request comes in, it appends the internal state to the request and emits
# a extended request
# Whenever a pageview comes in, the internal state is updated but nothing is emitted
class CountPageviewsStateful(beam.DoFn):
# The internal states
NUM_PAGEVIEWS = userstate.CombiningValueStateSpec('num_pageviews', combine_fn=sum)
FIRST_PAGEVIEW = userstate.ReadModifyWriteStateSpec('first_pageview', coder=beam.coders.VarIntCoder())
def process(self,
element,
num_pageviews_state=beam.DoFn.StateParam(NUM_PAGEVIEWS),
first_pageview_state=beam.DoFn.StateParam(FIRST_PAGEVIEW)
):
import datetime
# Extract element
session_id = element[0]
event_type, event = element[1]
# Process different event types
# Depending on event type, different actions are done
if event_type == 'request':
# This is a request
request = event
# First, the first pageview timestamp is extracted and the seconds since first timestamp are calculated
first_pageview = first_pageview_state.read()
if first_pageview is not None:
seconds_since_first_pageview = (int(request['timestamp_utc'].timestamp()) - first_pageview)
first_pageview_timestamp_utc = datetime.datetime.utcfromtimestamp(float(first_pageview))
first_pageview_timestamp = first_pageview_timestamp_utc.strftime("%Y-%m-%d %H:%M:%S")
else:
seconds_since_first_pageview = -1
first_pageview_timestamp = None
# The calculated data is appended to the request
request['num_pageviews'] = num_pageviews_state.read()
request['first_pageview_timestamp'] = first_pageview_timestamp
request['seconds_since_first_pageview'] = seconds_since_first_pageview
# The pageview counter is reset
num_pageviews_state.clear()
# The request is returned
yield (session_id, request)
elif event_type == 'pageview':
# This is a pageview
pageview = event
# Update first pageview state
first_pageview = first_pageview_state.read()
if first_pageview is None:
first_pageview_state.write(int(pageview['timestamp_utc'].timestamp()))
elif first_pageview > int(pageview['timestamp_utc'].timestamp()):
first_pageview_state.write(int(pageview['timestamp_utc'].timestamp()))
# Increase number of pageviews
num_pageviews_state.add(1)
# Do not return anything, pageviews are not further processed
# This DoFn logs the delay between the event time and the processing time
class LogTimeDelay(beam.DoFn):
def process(self, element, timestamp=beam.DoFn.TimestampParam):
import datetime
import logging
timestamp_utc = datetime.datetime.utcfromtimestamp(float(timestamp))
seconds_delay = (datetime.datetime.utcnow() - timestamp_utc).total_seconds()
logging.warning('Delayed by %s seconds', seconds_delay)
yield element
这似乎奏效了，让我在直接跑时平均延迟约1-2秒。在云数据流上，平均延迟约为0.5-1秒。总之，这似乎解决了问题的定义。
进一步考虑
不过，还有一些悬而未决的问题：
- 我使用全局窗口，这意味着就我而言，内部状态将永远保持。也许会话窗口是正确的方法：当x秒内没有页面浏览/请求时，窗口将关闭，内部状态将被释放
- 处理延迟有点高，但也许我需要稍微调整一下pubsub部分
- 我不知道这个解决方案比标准波束方法增加了多少开销或内存消耗。我也没有测试高工作量和并行化