Цикл в потоке данных - луч apache для операции - PullRequest
Новая
       13

Цикл в потоке данных - луч apache для операции

0 голосов
/ 28 января 2019

Я написал код для ежедневной пакетной операции в Appache Beam - код Python потока данных.Я пытаюсь сделать диапазон бегуна даты.то есть в настоящее время он работает нормально за вчерашний день - если я хочу запустить от начальной даты до конечной даты, это трудно. Просьба предложить какой-либо метод для этого.Пожалуйста, найдите мой фрагмент кода для вчерашнего запуска. 

start_date = '20180101'
end_date = '20190101'


p = beam.Pipeline(options=options)

read = (
        p
        | 'BQRead: ' >> BQReader(
    query=test_query.format(date=date))
)

transformed = (
        read
        | 'Transform 1 ' >> beam.ParDo(Transform1())
)

transformed | 'BQWrite' >> BQWriter(table + date, table_schema)

Я пробовал, как показано ниже, но он не работает 

start_date = datetime.strptime('20190101', "%Y%m%d")
end_date = datetime.strptime('20190110', "%Y%m%d")
dates = list(rrule.rrule(rrule.DAILY, dtstart=start_date, until=end_date))

for date in dates:
    ds_nd = date.strftime('%Y%m%d')

    p = beam.Pipeline(options=options)

    read = (
        p
        | 'BQRead: ' >> BQReader(
    query=test_query.format(date=ds_nd))
    )

    transformed = (
        read
        | 'Transform 1 ' >> beam.ParDo(Transform1())
    )

    transformed | 'BQWrite' >> BQWriter(table + ds_nd, table_schema)
...