If I have a pipeline running and I restart the taskmanager on which it's
executing the log shows - I find the "No restore state for
UnbounedSourceWrapper." interesting, as it seems to indicate that the
pipeline never stored a state in the first place?
Starting taskexecutor as a console application on host
flink-taskmanager-5d85dd6854-pm5bl.
2019-01-29 09:20:48,706 WARN org.apache.hadoop.util.NativeCodeLoader
- Unable to load native-hadoop library for your
platform... using builtin-java classes where applicable
2019-01-29 09:20:51,253 WARN
org.apache.flink.shaded.zookeeper.org.apache.zookeeper.ClientCnxn - SASL
configuration failed: javax.security.auth.login.LoginException: No JAAS
configuration section named 'Client' was found in specified JAAS
configuration file: '/tmp/jaas-7768141350028767113.conf'. Will continue
connection to Zookeeper server without SASL authentication, if Zookeeper
server allows it.
2019-01-29 09:20:51,281 ERROR
org.apache.flink.shaded.curator.org.apache.curator.ConnectionState -
Authentication failed
2019-01-29 09:21:53,814 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/View.AsSingleton/Combine.GloballyAsSingletonView/DropInputs/ParMultiDo(NoOp)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:53,828 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/MultiPartitionsWriteTables/Window.Into()/Window.Assign.out
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:53,834 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/SinglePartitionWriteTables/Window.Into()/Window.Assign.out
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:53,917 INFO
org.apache.beam.runners.flink.translation.wrappers.streaming.io.UnboundedSourceWrapper
- No restore state for UnbounedSourceWrapper.
2019-01-29 09:21:53,929 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/MultiPartitionsWriteTables/ParDo(GarbageCollectTemporaryFiles)/ParMultiDo(GarbageCollectTemporaryFiles)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:53,937 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/MultiPartitionsReshuffle/ReifyOriginalTimestamps/ParDo(Anonymous)/ParMultiDo(Anonymous)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:53,978 INFO
org.apache.beam.runners.flink.translation.wrappers.streaming.io.UnboundedSourceWrapper
- Unbounded Flink Source 0/1 is reading from sources:
[org.apache.beam.runners.core.construction.UnboundedReadFromBoundedSource$BoundedToUnboundedSourceAdapter@69908217
]
2019-01-29 09:21:54,002 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/MultiPartitionsReshuffle/Window.Into()/Window.Assign.out
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,008 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/View.AsSingleton/Combine.GloballyAsSingletonView/View.CreatePCollectionView/Combine.globally(Concatenate)/WithKeys/AddKeys/Map/ParMultiDo(Anonymous)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,011 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/TempFilePrefixView/Combine.GloballyAsSingletonView/View.CreatePCollectionView/Combine.globally(Concatenate)/WithKeys/AddKeys/Map/ParMultiDo(Anonymous)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,020 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/SinglePartitionWriteTables/ParDo(GarbageCollectTemporaryFiles)/ParMultiDo(GarbageCollectTemporaryFiles)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,080 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/TempFilePrefixView/Combine.GloballyAsSingletonView/View.CreatePCollectionView/Combine.globally(Concatenate)/Values/Values/Map/ParMultiDo(Anonymous)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,091 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/MultiPartitionsReshuffle/RestoreOriginalTimestamps/Reify.ExtractTimestampsFromValues/ParDo(Anonymous)/ParMultiDo(Anonymous)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,099 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/MultiPartitionsWriteTables/WithKeys/AddKeys/Map/ParMultiDo(Anonymous)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,107 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/View.AsSingleton/Combine.GloballyAsSingletonView/Combine.globally(Singleton)/WithKeys/AddKeys/Map/ParMultiDo(Anonymous)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,109 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/MultiPartitionsWriteTables/Values/Values/Map/ParMultiDo(Anonymous)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,119 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/SinglePartitionWriteTables/Values/Values/Map/ParMultiDo(Anonymous)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,118 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/TempFilePrefixView/Combine.GloballyAsSingletonView/View.VoidKeyToMultimapMaterialization/ParDo(VoidKeyToMultimapMaterialization)/ParMultiDo(VoidKeyToMultimapMaterialization)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,115 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/SinglePartitionsReshuffle/ReifyOriginalTimestamps/ParDo(Anonymous)/ParMultiDo(Anonymous)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,114 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/TempFilePrefixView/Combine.GloballyAsSingletonView/View.CreatePCollectionView/Combine.globally(Concatenate)/Combine.perKey(Concatenate)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,111 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/SinglePartitionWriteTables/WithKeys/AddKeys/Map/ParMultiDo(Anonymous)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,111 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/SinglePartitionsReshuffle/RestoreOriginalTimestamps/Reify.ExtractTimestampsFromValues/ParDo(Anonymous)/ParMultiDo(Anonymous)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,110 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/View.AsSingleton/Combine.GloballyAsSingletonView/View.CreatePCollectionView/Combine.globally(Concatenate)/Values/Values/Map/ParMultiDo(Anonymous)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,110 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/TempFilePrefixView/Combine.GloballyAsSingletonView/Combine.globally(Singleton)/WithKeys/AddKeys/Map/ParMultiDo(Anonymous)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,109 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/View.AsSingleton/Combine.GloballyAsSingletonView/View.VoidKeyToMultimapMaterialization/ParDo(VoidKeyToMultimapMaterialization)/ParMultiDo(VoidKeyToMultimapMaterialization)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,144 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/SinglePartitionsReshuffle/Window.Into()/Window.Assign.out
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,172 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/WriteGroupedRecords/ParMultiDo(WriteGroupedRecordsToFiles)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,176 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/View.AsSingleton/Combine.GloballyAsSingletonView/Combine.globally(Singleton)/Values/Values/Map/ParMultiDo(Anonymous)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,179 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/View.AsSingleton/Combine.GloballyAsSingletonView/View.CreatePCollectionView/Combine.globally(Concatenate)/Combine.perKey(Concatenate)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,189 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/View.AsSingleton/Combine.GloballyAsSingletonView/Combine.globally(Singleton)/Combine.perKey(Singleton)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,191 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/TempFilePrefixView/Combine.GloballyAsSingletonView/Combine.globally(Singleton)/Values/Values/Map/ParMultiDo(Anonymous)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,203 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/SinglePartitionsReshuffle/RestoreOriginalTimestamps/ReifyTimestamps.RemoveWildcard/ParDo(Anonymous)/ParMultiDo(Anonymous)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,210 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/TempFilePrefixView/Combine.GloballyAsSingletonView/Combine.globally(Singleton)/Combine.perKey(Singleton)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,217 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/MultiPartitionsReshuffle/RestoreOriginalTimestamps/ReifyTimestamps.RemoveWildcard/ParDo(Anonymous)/ParMultiDo(Anonymous)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,238 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/MultiPartitionsReshuffle/ExpandIterable/ParMultiDo(Anonymous)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,242 WARN org.apache.flink.metrics.MetricGroup
- The operator name
BigQueryIO.Write/BatchLoads/SinglePartitionsReshuffle/ExpandIterable/ParMultiDo(Anonymous)
exceeded the 80 characters length limit and was truncated.
2019-01-29 09:21:54,339 INFO
org.apache.beam.runners.flink.translation.wrappers.streaming.io.UnboundedSourceWrapper
- No restore state for UnbounedSourceWrapper.
2019-01-29 09:21:54,371 INFO
org.apache.beam.runners.flink.translation.wrappers.streaming.io.UnboundedSourceWrapper
- No restore state for UnbounedSourceWrapper.
2019-01-29 09:21:54,479 INFO
org.apache.beam.runners.flink.translation.wrappers.streaming.io.UnboundedSourceWrapper
- No restore state for UnbounedSourceWrapper.
2019-01-29 09:21:55,509 INFO
org.apache.beam.runners.flink.translation.wrappers.streaming.io.UnboundedSourceWrapper
- Unbounded Flink Source 0/1 is reading from sources:
[org.apache.beam.runners.core.construction.UnboundedReadFromBoundedSource$BoundedToUnboundedSourceAdapter@2a501a64
]
2019-01-29 09:21:55,535 INFO
org.apache.beam.runners.flink.translation.wrappers.streaming.io.UnboundedSourceWrapper
- Unbounded Flink Source 0/1 is reading from sources:
[org.apache.beam.runners.core.construction.UnboundedReadFromBoundedSource$BoundedToUnboundedSourceAdapter@60345813
]
2019-01-29 09:21:55,770 INFO
org.apache.beam.runners.flink.translation.wrappers.streaming.io.UnboundedSourceWrapper
- Unbounded Flink Source 0/1 is reading from sources:
[org.apache.beam.sdk.io.kafka.KafkaUnboundedSource@75aab48]
2019-01-29 09:21:56,280 WARN
org.apache.kafka.clients.consumer.ConsumerConfig - The
configuration 'metis.input.messages.config' was supplied but isn't a known
config.
2019-01-29 09:21:57,387 INFO
org.apache.beam.sdk.io.gcp.bigquery.BatchLoads - Writing
BigQuery temporary files to
gs://bucket/BigQueryWriteTemp/beam_load_ratingsflink01290921554a632112_85202e449b2d45729d384e3ac5f8cc2b/
before loading them.
2019-01-29 09:21:58,118 INFO
org.apache.beam.sdk.io.gcp.bigquery.BigQueryHelpers - Waiting for
jobs to complete.
2019-01-29 09:21:58,118 INFO
org.apache.beam.sdk.io.gcp.bigquery.BigQueryHelpers - Waiting for
jobs to complete.
2019-01-29 09:21:58,118 INFO
org.apache.beam.sdk.io.gcp.bigquery.BigQueryHelpers - Waiting for
jobs to complete.
2019-01-29 09:21:58,140 INFO
org.apache.beam.sdk.io.kafka.KafkaUnboundedSource - Reader-0:
reading from ratings-0 starting at offset 13112
2019-01-29 09:21:58,141 INFO
org.apache.beam.sdk.io.kafka.KafkaUnboundedSource - Reader-0:
reading from ratings-1 starting at offset 13407
2019-01-29 09:21:58,142 INFO
org.apache.beam.sdk.io.kafka.KafkaUnboundedSource - Reader-0:
reading from ratings-2 starting at offset 13034
2019-01-29 09:21:58,142 INFO
org.apache.beam.sdk.io.kafka.KafkaUnboundedSource - Reader-0:
reading from ratings-3 starting at offset 13271
2019-01-29 09:21:58,142 INFO
org.apache.beam.sdk.io.kafka.KafkaUnboundedSource - Reader-0:
reading from ratings-4 starting at offset 12813
2019-01-29 09:21:58,142 INFO
org.apache.beam.sdk.io.kafka.KafkaUnboundedSource - Reader-0:
reading from ratings-5 starting at offset 13211
2019-01-29 09:21:58,144 INFO
org.apache.beam.sdk.io.kafka.KafkaUnboundedSource - Reader-0:
reading from ratings-6 starting at offset 13394
2019-01-29 09:21:58,145 INFO
org.apache.beam.sdk.io.kafka.KafkaUnboundedSource - Reader-0:
reading from ratings-7 starting at offset 13194
2019-01-29 09:21:58,145 INFO
org.apache.beam.sdk.io.kafka.KafkaUnboundedSource - Reader-0:
reading from ratings-8 starting at offset 13478
2019-01-29 09:21:58,145 INFO
org.apache.beam.sdk.io.kafka.KafkaUnboundedSource - Reader-0:
reading from ratings-9 starting at offset 12966
On Mon, Jan 28, 2019 at 3:36 PM Kaymak, Tobias <[email protected]>
wrote:
> Hi Maximilian,
>
> yes, I've set the --runner to FlinkRunner when launching the pipeline and
> it does work for a GCS sink, but it seems to be ignored for a BigQuery sink
> somehow. Even though it looks like the system magically handles it itself.
>
> This is the full command line to launch the Beam 2.9.0 pipeline on Flink
> 1.5.5:
>
> bin/flink run -d -c di.beam.KafkaToBigQuery -j lib/beam_pipelines.jar
> --runner=FlinkRunner --appName=ratings --checkpointingMode=EXACTLY_ONCE
> --checkpointingInterval=300000 --parallelism=1
> --tempLocation=gs://somebucket
>
> Here are the logs from the taskmanager, I can share the full code of the
> pipeline if you want:
>
> 2019-01-28 14:33:31,287 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/View.AsSingleton/Combine.GloballyAsSingletonView/DropInputs/ParMultiDo(NoOp)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:31,911 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/SinglePartitionWriteTables/ParDo(GarbageCollectTemporaryFiles)/ParMultiDo(GarbageCollectTemporaryFiles)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:31,976 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/MultiPartitionsWriteTables/ParDo(GarbageCollectTemporaryFiles)/ParMultiDo(GarbageCollectTemporaryFiles)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:32,217 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/SinglePartitionWriteTables/Values/Values/Map/ParMultiDo(Anonymous)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:32,227 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/View.AsSingleton/Combine.GloballyAsSingletonView/Combine.globally(Singleton)/WithKeys/AddKeys/Map/ParMultiDo(Anonymous)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:32,228 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/SinglePartitionsReshuffle/RestoreOriginalTimestamps/Reify.ExtractTimestampsFromValues/ParDo(Anonymous)/ParMultiDo(Anonymous)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:32,276 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/MultiPartitionsWriteTables/Values/Values/Map/ParMultiDo(Anonymous)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:32,282 INFO
> org.apache.beam.runners.flink.translation.wrappers.streaming.io.UnboundedSourceWrapper
> - No restore state for UnbounedSourceWrapper.
> 2019-01-28 14:33:32,288 INFO
> org.apache.beam.runners.flink.translation.wrappers.streaming.io.UnboundedSourceWrapper
> - Unbounded Flink Source 0/1 is reading from sources:
> [org.apache.beam.runners.core.construction.UnboundedReadFromBoundedSource$BoundedToUnboundedSourceAdapter@5fecdf95
> ]
> 2019-01-28 14:33:32,296 INFO
> org.apache.beam.runners.flink.translation.wrappers.streaming.io.UnboundedSourceWrapper
> - No restore state for UnbounedSourceWrapper.
> 2019-01-28 14:33:32,318 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/TempFilePrefixView/Combine.GloballyAsSingletonView/View.CreatePCollectionView/Combine.globally(Concatenate)/WithKeys/AddKeys/Map/ParMultiDo(Anonymous)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:32,321 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/SinglePartitionsReshuffle/RestoreOriginalTimestamps/ReifyTimestamps.RemoveWildcard/ParDo(Anonymous)/ParMultiDo(Anonymous)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:32,324 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/TempFilePrefixView/Combine.GloballyAsSingletonView/View.CreatePCollectionView/Combine.globally(Concatenate)/Values/Values/Map/ParMultiDo(Anonymous)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:32,329 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/TempFilePrefixView/Combine.GloballyAsSingletonView/View.CreatePCollectionView/Combine.globally(Concatenate)/Combine.perKey(Concatenate)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:32,357 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/TempFilePrefixView/Combine.GloballyAsSingletonView/View.VoidKeyToMultimapMaterialization/ParDo(VoidKeyToMultimapMaterialization)/ParMultiDo(VoidKeyToMultimapMaterialization)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:32,482 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/SinglePartitionsReshuffle/ExpandIterable/ParMultiDo(Anonymous)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:32,483 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/TempFilePrefixView/Combine.GloballyAsSingletonView/Combine.globally(Singleton)/Values/Values/Map/ParMultiDo(Anonymous)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:32,493 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/TempFilePrefixView/Combine.GloballyAsSingletonView/Combine.globally(Singleton)/Combine.perKey(Singleton)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:32,697 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/WriteGroupedRecords/ParMultiDo(WriteGroupedRecordsToFiles)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:32,782 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/View.AsSingleton/Combine.GloballyAsSingletonView/View.CreatePCollectionView/Combine.globally(Concatenate)/Values/Values/Map/ParMultiDo(Anonymous)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:32,789 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/View.AsSingleton/Combine.GloballyAsSingletonView/View.CreatePCollectionView/Combine.globally(Concatenate)/Combine.perKey(Concatenate)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:33,093 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/View.AsSingleton/Combine.GloballyAsSingletonView/View.CreatePCollectionView/Combine.globally(Concatenate)/WithKeys/AddKeys/Map/ParMultiDo(Anonymous)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:33,122 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/View.AsSingleton/Combine.GloballyAsSingletonView/View.VoidKeyToMultimapMaterialization/ParDo(VoidKeyToMultimapMaterialization)/ParMultiDo(VoidKeyToMultimapMaterialization)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:33,162 INFO
> org.apache.beam.runners.flink.translation.wrappers.streaming.io.UnboundedSourceWrapper
> - No restore state for UnbounedSourceWrapper.
> 2019-01-28 14:33:33,179 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/MultiPartitionsReshuffle/RestoreOriginalTimestamps/Reify.ExtractTimestampsFromValues/ParDo(Anonymous)/ParMultiDo(Anonymous)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:33,187 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/MultiPartitionsReshuffle/RestoreOriginalTimestamps/ReifyTimestamps.RemoveWildcard/ParDo(Anonymous)/ParMultiDo(Anonymous)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:33,192 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/MultiPartitionsReshuffle/ExpandIterable/ParMultiDo(Anonymous)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:33,218 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/View.AsSingleton/Combine.GloballyAsSingletonView/Combine.globally(Singleton)/Values/Values/Map/ParMultiDo(Anonymous)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:33,220 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/View.AsSingleton/Combine.GloballyAsSingletonView/Combine.globally(Singleton)/Combine.perKey(Singleton)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:33,298 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/SinglePartitionWriteTables/Window.Into()/Window.Assign.out
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:33,304 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/MultiPartitionsWriteTables/Window.Into()/Window.Assign.out
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:33,323 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/MultiPartitionsWriteTables/WithKeys/AddKeys/Map/ParMultiDo(Anonymous)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:33,326 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/MultiPartitionsReshuffle/ReifyOriginalTimestamps/ParDo(Anonymous)/ParMultiDo(Anonymous)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:33,357 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/SinglePartitionWriteTables/WithKeys/AddKeys/Map/ParMultiDo(Anonymous)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:33,377 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/MultiPartitionsReshuffle/Window.Into()/Window.Assign.out
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:33,395 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/TempFilePrefixView/Combine.GloballyAsSingletonView/Combine.globally(Singleton)/WithKeys/AddKeys/Map/ParMultiDo(Anonymous)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:33,477 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/SinglePartitionsReshuffle/ReifyOriginalTimestamps/ParDo(Anonymous)/ParMultiDo(Anonymous)
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:33,487 WARN org.apache.flink.metrics.MetricGroup
> - The operator name
> BigQueryIO.Write/BatchLoads/SinglePartitionsReshuffle/Window.Into()/Window.Assign.out
> exceeded the 80 characters length limit and was truncated.
> 2019-01-28 14:33:33,748 INFO
> org.apache.beam.runners.flink.translation.wrappers.streaming.io.UnboundedSourceWrapper
> - No restore state for UnbounedSourceWrapper.
> 2019-01-28 14:33:34,577 INFO
> org.apache.beam.runners.flink.translation.wrappers.streaming.io.UnboundedSourceWrapper
> - Unbounded Flink Source 0/1 is reading from sources:
> [org.apache.beam.runners.core.construction.UnboundedReadFromBoundedSource$BoundedToUnboundedSourceAdapter@1892a35d
> ]
> 2019-01-28 14:33:34,610 INFO
> org.apache.beam.runners.flink.translation.wrappers.streaming.io.UnboundedSourceWrapper
> - Unbounded Flink Source 0/1 is reading from sources:
> [org.apache.beam.runners.core.construction.UnboundedReadFromBoundedSource$BoundedToUnboundedSourceAdapter@7adcb05b
> ]
> 2019-01-28 14:33:34,747 INFO
> org.apache.beam.runners.flink.translation.wrappers.streaming.io.UnboundedSourceWrapper
> - Unbounded Flink Source 0/1 is reading from sources:
> [org.apache.beam.sdk.io.kafka.KafkaUnboundedSource@71389814]
> 2019-01-28 14:33:34,896 WARN
> org.apache.kafka.clients.consumer.ConsumerConfig - The
> configuration 'metis.input.messages.config' was supplied but isn't a known
> config.
> 2019-01-28 14:33:35,462 INFO
> org.apache.beam.sdk.io.gcp.bigquery.BatchLoads - Writing
> BigQuery temporary files to
> gs://tempspace_eu_regional/dataflow_dev/BigQueryWriteTemp/beam_load_ratingsflink01281433342cd8f9b4_7ce229196b1d41e3ad63d0ef6234e0f6/
> before loading them.
> 2019-01-28 14:33:35,544 INFO
> org.apache.beam.sdk.io.kafka.KafkaUnboundedSource - Reader-0:
> reading from ratings-0 starting at offset 2945
> 2019-01-28 14:33:35,544 INFO
> org.apache.beam.sdk.io.kafka.KafkaUnboundedSource - Reader-0:
> reading from ratings-1 starting at offset 3101
> 2019-01-28 14:33:35,544 INFO
> org.apache.beam.sdk.io.kafka.KafkaUnboundedSource - Reader-0:
> reading from ratings-2 starting at offset 3031
> 2019-01-28 14:33:35,545 INFO
> org.apache.beam.sdk.io.kafka.KafkaUnboundedSource - Reader-0:
> reading from ratings-3 starting at offset 3009
> 2019-01-28 14:33:35,545 INFO
> org.apache.beam.sdk.io.kafka.KafkaUnboundedSource - Reader-0:
> reading from ratings-4 starting at offset 2903
> 2019-01-28 14:33:35,545 INFO
> org.apache.beam.sdk.io.kafka.KafkaUnboundedSource - Reader-0:
> reading from ratings-5 starting at offset 3068
> 2019-01-28 14:33:35,545 INFO
> org.apache.beam.sdk.io.kafka.KafkaUnboundedSource - Reader-0:
> reading from ratings-6 starting at offset 3160
> 2019-01-28 14:33:35,545 INFO
> org.apache.beam.sdk.io.kafka.KafkaUnboundedSource - Reader-0:
> reading from ratings-7 starting at offset 3014
> 2019-01-28 14:33:35,546 INFO
> org.apache.beam.sdk.io.kafka.KafkaUnboundedSource - Reader-0:
> reading from ratings-8 starting at offset 3096
> 2019-01-28 14:33:35,546 INFO
> org.apache.beam.sdk.io.kafka.KafkaUnboundedSource - Reader-0:
> reading from ratings-9 starting at offset 2885
> 2019-01-28 14:33:35,577 WARN
> org.apache.kafka.clients.consumer.ConsumerConfig - The
> configuration 'metis.input.messages.config' was supplied but isn't a known
> config.
> 2019-01-28 14:33:35,801 INFO
> org.apache.beam.sdk.io.gcp.bigquery.BigQueryHelpers - Waiting for
> jobs to complete.
> 2019-01-28 14:33:35,803 INFO
> org.apache.beam.sdk.io.gcp.bigquery.BigQueryHelpers - Waiting for
> jobs to complete.
> 2019-01-28 14:33:35,801 INFO
> org.apache.beam.sdk.io.gcp.bigquery.BigQueryHelpers - Waiting for
> jobs to complete.
> 2019-01-28 14:33:36,217 INFO
> org.apache.beam.sdk.io.kafka.KafkaUnboundedSource - Reader-0:
> first record offset 3014
>
>
> Best,
> Tobi
>
>
> On Mon, Jan 28, 2019 at 11:52 AM Maximilian Michels <[email protected]>
> wrote:
>
>> Hi Tobias,
>>
>> Checkpointing should be enabled when you set it in the Flink config or
>> via the
>> Beam option `checkpointingInterval`. Did you set `runner` to
>> `FlinkRunner`?
>>
>> If possible, could you share parts of the Flink logs?
>>
>> Thanks,
>> Max
>>
>> On 25.01.19 15:14, Kaymak, Tobias wrote:
>> > Hi,
>> >
>> > I am trying to migrate my existing KafkaToGCS pipeline to a
>> KafkaToBigQuery
>> > pipeline to skip the loading step from GCS which is currently handled
>> externally
>> > from Beam.
>> >
>> > I noticed that the pipeline, written in Beam 2.9.0 (Java) does not
>> trigger any
>> > checkpoint on Flink (1.5.5), even though its configured to do so when I
>> launch
>> > it. Is this normal? How does Beam then guarantee exactly once when
>> there are no
>> > checkpoints in Flink? (It seems to start from scratch when it crashes,
>> during my
>> > tests, but I am not 100% sure)
>> >
>> >
>> > This is my pipeline:
>> >
>> > pipeline
>> > .apply(
>> > KafkaIO.<String, String>read()
>> > .withBootstrapServers(bootstrap)
>> > .withTopics(topics)
>> > .withKeyDeserializer(StringDeserializer.class)
>> > .withValueDeserializer(ConfigurableDeserializer.class)
>> > .updateConsumerProperties(
>> >
>> ImmutableMap.of(InputMessagesConfig.CONFIG_PROPERTY_NAME,
>> > inputMessagesConfig))
>> >
>> .updateConsumerProperties(ImmutableMap.of("auto.offset.reset",
>> > "earliest"))
>> > .updateConsumerProperties(ImmutableMap.of("group.id
>> > <http://group.id>", groupId))
>> >
>> .updateConsumerProperties(ImmutableMap.of("enable.auto.commit",
>> > "true"))
>> > .withReadCommitted()
>> > .withTimestampPolicyFactory(withEventTs)
>> > .commitOffsetsInFinalize())
>> > .apply(ParDo.of(new ToEventFn()))
>> > .apply(
>> > Window.into(new ZurichTimePartitioningWindowFn())
>> >
>> > .triggering(
>> > Repeatedly.forever(
>> > AfterFirst.of(
>> > AfterPane.elementCountAtLeast(bundleSize),
>> >
>> AfterProcessingTime.pastFirstElementInPane()
>> > .plusDelayOf(refreshFrequency))))
>> > .withAllowedLateness(Duration.standardDays(14))
>> > .discardingFiredPanes())
>> > .apply(
>> > BigQueryIO.<Event>write()
>> > .withMethod(BigQueryIO.Write.Method.FILE_LOADS)
>> > .withTriggeringFrequency(refreshFrequency)
>> > .withNumFileShards(1)
>> > .to(partitionedTableDynamicDestinations)
>> > .withFormatFunction(
>> > (SerializableFunction<Event, TableRow>)
>> > KafkaToBigQuery::convertUserEventToTableRow)
>> >
>> >
>> .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
>> >
>> > .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND));
>> >
>> > pipeline.run().waitUntilFinish();
>> > It's launched like the other (GCS) one via:
>> >
>> > ...--checkpointingMode=EXACTLY_ONCE --checkpointingInterval=300000
>> > --parallelism=1 --tempLocation=gs://foo..
>> >
>> > Any idea why checkpointing does not work here?
>> >
>> > Best,
>> > Tobias
>>
>
>
> --
> Tobias Kaymak
> Data Engineer
>
> [email protected]
> www.ricardo.ch
>
--
Tobias Kaymak
Data Engineer
[email protected]
www.ricardo.ch
