apache / apache/beam

beam.Create([...]) does not create an properly shaped PCollection until pipeline executes

Open
#19,641 0 comments 0 reactions 0 assignees View on GitHub
core improvement P3 python
Dominant language
Java
Stars
8.7k
Forks
4.7k
Avg merge
1d 20h
Merged PRs (30d)
196

Description

This issue is surfaced by the errors from [https://github.com/apache/beam/pull/9077](https://github.com/apache/beam/pull/9077)

 

Specifically, in the following pipelines, note how Map(lambda x: x) is necessary before the actual transform that is being tested. This is due to the fact that the pcoll output by create is a little odd.
```

// code placeholder
def test_timestamp(self):
l = [TimestampedValue('a', 100),
TimestampedValue('b',
200),
TimestampedValue('c', 300)]
expected = [TestWindowedValue('a', 100, [GlobalWindow()]),

TestWindowedValue('b', 200, [GlobalWindow()]),
TestWindowedValue('c',
300, [GlobalWindow()])]
with TestPipeline() as p:
# Map(lambda x: x) PTransform is added
after Create here, because when
# a PCollection of TimestampedValues is created with Create PTransform,

# the timestamps are not assigned to it. Adding a Map forces the
# PCollection to go through
a DoFn so that the PCollection consists of
# the elements with timestamps assigned to them instead
of a PCollection
# of TimestampedValue(element, timestamp).
pc = p | beam.Create(l) |
beam.Map(lambda x: x)
reified_pc = pc | util.Reify.Timestamp()
assert_that(reified_pc,
equal_to(expected), reify_windows=True)

def test_window(self):
l = [GlobalWindows.windowed_value('a',
100),
GlobalWindows.windowed_value('b', 200),
GlobalWindows.windowed_value('c',
300)]
expected = [TestWindowedValue(('a', 100, GlobalWindow()), 100,

[GlobalWindow()]),
TestWindowedValue(('b', 200, GlobalWindow()), 200,

[GlobalWindow()]),
TestWindowedValue(('c', 300, GlobalWindow()),
300,
[GlobalWindow()])]
with TestPipeline() as p:
pc
= p | beam.Create(l)
# Map(lambda x: x) PTransform is added after Create here, because when

# a PCollection of WindowedValues is created with Create PTransform,
# the windows are not
assigned to it. Adding a Map forces the
# PCollection to go through a DoFn so that the PCollection
consists of
# the elements with timestamps assigned to them instead of a PCollection
#
of WindowedValue(element, timestamp, window).
pc = pc | beam.Map(lambda x: x)
reified_pc
= pc | util.Reify.Window()
assert_that(reified_pc, equal_to(expected), reify_windows=True)

```

 

Imported from Jira [BEAM-7759](https://issues.apache.org/jira/browse/BEAM-7759). Original Jira may contain additional context.
Reported by: pabloem.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.