ankitkumarrain commented on issue #35934:
URL: https://github.com/apache/beam/issues/35934#issuecomment-3354903614
Hello everyone here , i have written code is it mergable . import
apache_beam as beam
from apache_beam.transforms import core
from apache_beam.transforms.window import GlobalWindows, WindowInto,
Repeatedly, AfterCount, AccumulationMode
from apache_beam.transforms.periodicsequence import PeriodicImpulse
from apache_beam.transforms.combiners import Latest
import logging
import time
pipeline = beam.Pipeline(options=options)
start_timestamp = time.time()
main_input_fire_interval = 60
side_input_fire_interval = 60
def add_timestamp(element, timestamp=core.DoFn.TimestampParam):
return (element, timestamp)
side_input = (
pipeline
| "SideInputImpulse" >> PeriodicImpulse(
start_timestamp=start_timestamp,
fire_interval=main_input_fire_interval,
)
| "SideInputWindow" >> WindowInto(
GlobalWindows(),
trigger=Repeatedly(AfterCount(1)),
accumulation_mode=AccumulationMode.DISCARDING,
)
| "Add timestamp" >> core.ParDo(add_timestamp)
)
(
side_input
| "LocalLatest" >> Latest.Globally().without_defaults()
| "Show latest" >> beam.Map(logging.info)
)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
