--- title: Google Cloud Dataflow description: >- A managed service for transforming and enriching data in both real-time and historical modes. breadcrumbs: Docs > Integrations > Google Cloud Dataflow --- > For the complete documentation index, see [llms.txt](https://docs.datadoghq.com/llms.txt). # Google Cloud Dataflow {% callout %} # Important note for users on the following Datadog sites: us2.ddog-gov.com {% alert level="info" %} To find out if this integration is available in your organization, see your [Datadog Integrations](https://app.datadoghq.com/integrations) page or ask your organization administrator. To initiate an exception request to enable this integration for your organization, email [support@ddog-gov.com](mailto:support@ddog-gov.com). {% /alert %} {% /callout %} ## Overview{% #overview %} Google Cloud Dataflow is a fully-managed service for transforming and enriching data in stream (real time) and batch (historical) modes with equal reliability and expressiveness. Use the Datadog Google Cloud integration to collect metrics from Google Cloud Dataflow. ## Setup{% #setup %} ### Metric collection{% #metric-collection %} #### Installation{% #installation %} If you haven't already, set up the [Google Cloud Platform integration](https://docs.datadoghq.com/integrations/google-cloud-platform.md) first. There are no other installation steps. ### Log collection{% #log-collection %} Google Cloud Dataflow logs are collected with Google Cloud Logging and sent to a Dataflow job through a Cloud Pub/Sub topic. If you haven't already, [set up logging with the Datadog Dataflow template](https://docs.datadoghq.com/integrations/google-cloud-platform.md?tab=datadogussite#log-collection). Once this is done, export your Google Cloud Dataflow logs from Google Cloud Logging to the Pub/sub: 1. Go to the [Google Cloud Logging page](https://console.cloud.google.com/logs/viewer) and filter the Google Cloud Dataflow logs. 1. Click **Create Sink** and name the sink accordingly. 1. Choose "Cloud Pub/Sub" as the destination and select the Pub/Sub topic that was created for that purpose. **Note**: The Pub/Sub topic can be located in a different project. 1. Click **Create** and wait for the confirmation message to show up. ## Data Collected{% #data-collected %} ### Metrics{% #metrics %} | | | | | **gcp.dataflow.job.active_worker_instances**(gauge) | The active number of worker instances. Streaming engine jobs only. | | **gcp.dataflow.job.aggregated_worker_utilization**(gauge) | Aggregated worker utilization (e.g. worker CPU utilization) across the worker pool.*Shown as percent* | | **gcp.dataflow.job.backlog_bytes**(gauge) | Amount of known, unprocessed input for a stage, in bytes.*Shown as byte* | | **gcp.dataflow.job.backlog_elements**(gauge) | Amount of known, unprocessed input for a stage, in elements. | | **gcp.dataflow.job.bigquery.write_count**(count) | BigQuery write requests from BigQueryIO.Write in Dataflow jobs. | | **gcp.dataflow.job.billable_shuffle_data_processed**(gauge) | The billable bytes of shuffle data processed by this Dataflow job.*Shown as byte* | | **gcp.dataflow.job.bundle_user_processing_latencies.avg**(gauge) | Bundle user processing latencies from a particular stage. Available for jobs running on Streaming Engine.*Shown as millisecond* | | **gcp.dataflow.job.bundle_user_processing_latencies.samplecount**(gauge) | Bundle user processing latencies from a particular stage. Available for jobs running on Streaming Engine. | | **gcp.dataflow.job.bundle_user_processing_latencies.sumsqdev**(gauge) | Bundle user processing latencies from a particular stage. Available for jobs running on Streaming Engine. | | **gcp.dataflow.job.current_num_vcpus**(gauge) | The number of vCPUs currently being used by this Dataflow job.*Shown as cpu* | | **gcp.dataflow.job.current_shuffle_slots**(gauge) | The current shuffle slots used by this Dataflow job. | | **gcp.dataflow.job.data_watermark_age**(gauge) | The age (time since event timestamp) of the most recent item of data that has been fully processed by the pipeline.*Shown as second* | | **gcp.dataflow.job.disk_space_capacity**(gauge) | The amount of persistent disk currently being allocated to all workers associated with this Dataflow job.*Shown as byte* | | **gcp.dataflow.job.dofn_latency_average**(gauge) | The average processing time for a single message in a given DoFn (over the past 3 min window). Note that this includes time spent in GetData calls. Available for jobs running on Streaming Engine on the Legacy Runner.*Shown as millisecond* | | **gcp.dataflow.job.dofn_latency_max**(gauge) | The maximum processing time for a single message in a given DoFn (over the past 3 min window). Note that this includes time spent in GetData calls. Available for jobs running on Streaming Engine on the Legacy Runner.*Shown as millisecond* | | **gcp.dataflow.job.dofn_latency_min**(gauge) | The minimum processing time for a single message in a given DoFn (over the past 3 min window). Available for jobs running on Streaming Engine on the Legacy Runner.*Shown as millisecond* | | **gcp.dataflow.job.dofn_latency_num_messages**(gauge) | The number of messages processed by a given DoFn (over the past 3 min window). Available for jobs running on Streaming Engine on the Legacy Runner. | | **gcp.dataflow.job.dofn_latency_total**(gauge) | The total processing time for all messages in a given DoFn (over the past 3 min window). Note that this includes time spent in GetData calls. Available for jobs running on Streaming Engine on the Legacy Runner.*Shown as millisecond* | | **gcp.dataflow.job.duplicates_filtered_out_count**(count) | The number of messages being processed by a particular stage that have been filtered out as duplicates. Available for jobs running on Streaming Engine. | | **gcp.dataflow.job.elapsed_time**(gauge) | Duration that the current run of this pipeline has been in the Running state so far, in seconds. When a run completes, this stays at the duration of that run until the next run starts.*Shown as second* | | **gcp.dataflow.job.element_count**(count) | Number of elements added to the PCollection so far.*Shown as item* | | **gcp.dataflow.job.elements_produced_count**(count) | The number of elements produced by each PTransform. | | **gcp.dataflow.job.estimated_backlog_processing_time**(gauge) | Estimated time (in seconds) to consume current backlog if no new data comes in and throughput stays the same. Only available for Streaming Engine jobs.*Shown as second* | | **gcp.dataflow.job.estimated_byte_count**(count) | An estimated number of bytes added to the PCollection so far.*Shown as byte* | | **gcp.dataflow.job.estimated_bytes_active**(gauge) | Estimated number of bytes active in this stage of the job.*Shown as byte* | | **gcp.dataflow.job.estimated_bytes_consumed_count**(count) | Estimated number of bytes consumed by the stage of this job.*Shown as byte* | | **gcp.dataflow.job.estimated_bytes_produced_count**(count) | The estimated total byte size of elements produced by each PTransform. | | **gcp.dataflow.job.estimated_timer_backlog_processing_time**(gauge) | Estimated time (in seconds) for timers to complete. Only available for Streaming Engine jobs.*Shown as second* | | **gcp.dataflow.job.gpu_memory_utilization**(gauge) | Percent of time over the past sample period during which global (device) memory was being read or written.*Shown as percent* | | **gcp.dataflow.job.gpu_utilization**(gauge) | Percent of time over the past sample period during which one or more kernels was executing on the GPU.*Shown as percent* | | **gcp.dataflow.job.horizontal_worker_scaling**(gauge) | A boolean value that indicates what kind of horizontal scaling direction which the autoscaler recommended and rationale behind it. A true metric output means a scaling decision is made and a false metric output means the corresponding scaling is not taking effect. | | **gcp.dataflow.job.is_failed**(gauge) | Has this job failed. | | **gcp.dataflow.job.max_worker_instances_limit**(gauge) | The maximum number of workers autoscaling is allowed to request. | | **gcp.dataflow.job.memory_capacity**(gauge) | The amount of memory currently being allocated to all workers associated with this Dataflow job.*Shown as byte* | | **gcp.dataflow.job.min_worker_instances_limit**(gauge) | The minimum number of workers autoscaling is allowed to request. | | **gcp.dataflow.job.oldest_active_message_age**(gauge) | How long the oldest active message in a DoFn has been processing for. Available for jobs running on Streaming Engine on the Legacy Runner.*Shown as millisecond* | | **gcp.dataflow.job.per_stage_data_watermark_age**(gauge) | The age (time since event timestamp) up to which all data has been processed by this stage of the pipeline.*Shown as second* | | **gcp.dataflow.job.per_stage_system_lag**(gauge) | The current maximum duration that an item of data has been processing or awaiting processing in seconds, per pipeline stage.*Shown as second* | | **gcp.dataflow.job.processing_parallelism_keys**(gauge) | Approximate number of keys in use for data processing for each stage. Processing for any given key is serialized, so the total number of keys for a stage represents the maximum available parallelism at that stage. Available for jobs running on Streaming Engine. | | **gcp.dataflow.job.pubsub.late_messages_count**(count) | The number of messages from Pub/Sub with timestamp older than the estimated watermark. | | **gcp.dataflow.job.pubsub.published_messages_count**(count) | The number of Pub/Sub messages published broken down by topic and status. | | **gcp.dataflow.job.pubsub.pulled_message_ages.avg**(gauge) | The distribution of pulled but unacked Pub/Sub message ages.*Shown as millisecond* | | **gcp.dataflow.job.pubsub.pulled_message_ages.samplecount**(gauge) | The distribution of pulled but unacked Pub/Sub message ages. | | **gcp.dataflow.job.pubsub.pulled_message_ages.sumsqdev**(gauge) | The distribution of pulled but unacked Pub/Sub message ages. | | **gcp.dataflow.job.pubsub.read_count**(count) | Pub/Sub Pull Requests. For Streaming Engine, this metric is deprecated. See the "Using the Dataflow monitoring interface" page for upcoming changes. | | **gcp.dataflow.job.pubsub.read_latencies.avg**(count) | The average Pub/Sub Pull request latencies from PubsubIO.Read in Dataflow jobs. For Streaming Engine, this metric is deprecated. See the [Using the Dataflow monitoring interface](https://cloud.google.com/dataflow/docs/guides/monitoring-overview) page for upcoming changes.*Shown as millisecond* | | **gcp.dataflow.job.pubsub.read_latencies.samplecount**(count) | Pub/Sub Pull request latencies from PubsubIO.Read in Dataflow jobs. For Streaming Engine, this metric is deprecated. See the "Using the Dataflow monitoring interface" page for upcoming changes. | | **gcp.dataflow.job.pubsub.read_latencies.sumsqdev**(count) | Pub/Sub Pull request latencies from PubsubIO.Read in Dataflow jobs. For Streaming Engine, this metric is deprecated. See the "Using the Dataflow monitoring interface" page for upcoming changes. | | **gcp.dataflow.job.pubsub.streaming_pull_connection_status**(gauge) | Percentage of all Streaming Pull connections that are either active (OK status) or terminated because of an error (non-OK status). When a connection is terminated, Dataflow will wait some time before attempting to re-connect. For Streaming Engine only.*Shown as percent* | | **gcp.dataflow.job.pubsub.write_count**(count) | Pub/Sub Publish requests from PubsubIO.Write in Dataflow jobs. | | **gcp.dataflow.job.pubsub.write_latencies.avg**(count) | The average Pub/Sub Publish request latencies from PubsubIO.Write in Dataflow jobs.*Shown as millisecond* | | **gcp.dataflow.job.pubsub.write_latencies.samplecount**(count) | Pub/Sub Publish request latencies from PubsubIO.Write in Dataflow jobs. | | **gcp.dataflow.job.pubsub.write_latencies.sumsqdev**(count) | Pub/Sub Publish request latencies from PubsubIO.Write in Dataflow jobs. | | **gcp.dataflow.job.status**(gauge) | Current state of this pipeline (e.g.: RUNNING, DONE, CANCELLED, FAILED, …). Not reported while the pipeline is not running. | | **gcp.dataflow.job.streaming_engine.key_processing_availability**(gauge) | Percentage of streaming processing keys that are assigned to workers and available to perform work. Work for unavailable keys will be deferred until keys are available.*Shown as percent* | | **gcp.dataflow.job.streaming_engine.persistent_state.read_bytes_count**(count) | Storage bytes read by a particular stage. Available for jobs running on Streaming Engine. | | **gcp.dataflow.job.streaming_engine.persistent_state.stored_bytes**(gauge) | Current bytes stored in persistent state for the job.*Shown as byte* | | **gcp.dataflow.job.streaming_engine.persistent_state.write_bytes_count**(count) | Storage bytes written by a particular stage. Available for jobs running on Streaming Engine. | | **gcp.dataflow.job.streaming_engine.persistent_state.write_latencies.avg**(count) | The average storage write latencies from a particular stage. Available for jobs running on Streaming Engine.*Shown as millisecond* | | **gcp.dataflow.job.streaming_engine.persistent_state.write_latencies.samplecount**(count) | Storage write latencies from a particular stage. Available for jobs running on Streaming Engine. | | **gcp.dataflow.job.streaming_engine.persistent_state.write_latencies.sumsqdev**(count) | Storage write latencies from a particular stage. Available for jobs running on Streaming Engine. | | **gcp.dataflow.job.streaming_engine.stage_end_to_end_latencies.avg**(gauge) | Distribution of time spent by streaming engine in each stage of the pipeline. This time includes shuffling messages, queueing them for processing, processing, queueing for persistent state write, and the write itself.*Shown as millisecond* | | **gcp.dataflow.job.streaming_engine.stage_end_to_end_latencies.samplecount**(gauge) | Distribution of time spent by streaming engine in each stage of the pipeline. This time includes shuffling messages, queueing them for processing, processing, queueing for persistent state write, and the write itself. | | **gcp.dataflow.job.streaming_engine.stage_end_to_end_latencies.sumsqdev**(gauge) | Distribution of time spent by streaming engine in each stage of the pipeline. This time includes shuffling messages, queueing them for processing, processing, queueing for persistent state write, and the write itself. | | **gcp.dataflow.job.system_lag**(gauge) | The current maximum duration that an item of data has been awaiting processing, in seconds.*Shown as second* | | **gcp.dataflow.job.target_worker_instances**(gauge) | The desired number of worker instances. | | **gcp.dataflow.job.thread_time**(count) | Estimated time in milliseconds spent running in the function of the ptransform totaled across threads on all workers of the job.*Shown as millisecond* | | **gcp.dataflow.job.timers_pending_count**(count) | The number of timers pending in a particular stage. Available for jobs running on Streaming Engine. | | **gcp.dataflow.job.timers_processed_count**(count) | The number of timers completed by a particular stage. Available for jobs running on Streaming Engine. | | **gcp.dataflow.job.total_dcu_usage**(count) | The total amount of DCUs (Data Compute Unit) used by the Dataflow job since it was launched. | | **gcp.dataflow.job.total_memory_usage_time**(gauge) | The total GB seconds of memory allocated to this Dataflow job.*Shown as gibibyte* | | **gcp.dataflow.job.total_pd_usage_time**(gauge) | The total GB seconds for all persistent disk used by all workers associated with this Dataflow job.*Shown as gibibyte* | | **gcp.dataflow.job.total_secu_usage**(gauge) | The total amount of SECUs (Streaming Engine Compute Unit) used by the Dataflow job since it was launched. | | **gcp.dataflow.job.total_shuffle_data_processed**(gauge) | The total bytes of shuffle data processed by this Dataflow job.*Shown as byte* | | **gcp.dataflow.job.total_streaming_data_processed**(gauge) | The total bytes of streaming data processed by this Dataflow job.*Shown as byte* | | **gcp.dataflow.job.total_vcpu_time**(gauge) | The total vCPU seconds used by this Dataflow job. | | **gcp.dataflow.job.user_counter**(gauge) | A user-defined counter metric. | | **gcp.dataflow.job.worker_utilization_hint**(gauge) | Worker utilization hint for autoscaling. This hint value is configured by the customers and defines a target worker CPU utilization range thus influencing scaling aggressiveness.*Shown as percent* | | **gcp.dataflow.job.worker_utilization_hint_is_actively_used**(gauge) | Reports whether or not the worker utilization hint is actively used by the horizontal autoscaling policy. | | **gcp.dataflow.quota.region_endpoint_shuffle_slot.exceeded**(count) | Number of attempts to exceed the limit on quota metric dataflow.googleapis.com/region_endpoint_shuffle_slot. After sampling, data is not visible for up to 150 seconds. | | **gcp.dataflow.quota.region_endpoint_shuffle_slot.limit**(gauge) | Current limit on quota metric `dataflow.googleapis.com/region_endpoint_shuffle_slot`. | | **gcp.dataflow.quota.region_endpoint_shuffle_slot.usage**(gauge) | Current usage on quota metric dataflow.googleapis.com/region_endpoint_shuffle_slot. After sampling, data is not visible for up to 150 seconds. | | **gcp.dataflow.worker.memory.bytes_used**(gauge) | The memory usage in bytes by a particular container instance on a Dataflow worker.*Shown as byte* | | **gcp.dataflow.worker.memory.container_limit**(gauge) | Maximum RAM size in bytes available to a particular container instance on a Dataflow worker.*Shown as byte* | | **gcp.dataflow.worker.memory.total_limit**(gauge) | RAM size in bytes available to a Dataflow worker.*Shown as byte* | | **gcp.dataflow.job.archived_shuffle_data**(gauge) | The archived shuffle bytes incurring storage cost, at any time the job is paused.*Shown as byte* | | **gcp.dataflow.job.backlogged_keys**(gauge) | The number of backlogged keys for a bottleneck stage. If a stage has backlogged keys, then it signifies its inability to process data at the required rate and the stage could be a bottleneck that slows down the overall job's progress. | | **gcp.dataflow.job.is_bottleneck**(gauge) | Whether a specific Dataflow pipeline stage is a bottleneck, along with its bottleneck kind and likely cause. See the troubleshooting guide [https://cloud.google.com/dataflow/docs/guides/troubleshoot-bottlenecks](https://cloud.google.com/dataflow/docs/guides/troubleshoot-bottlenecks) for more details. | | **gcp.dataflow.job.recommended_parallelism**(gauge) | The recommended parallelism for a stage to reduce bottlenecking. If this is a source stage, then the parallelism is a consequence of source partition count and the source may need to be repartitioned to increase parallelism. | | **gcp.dataflow.job.total_billable_shuffle_archived_data_time**(gauge) | The total gigabyte seconds of shuffle archived data. This shows the cumulative cost accrued from all past sessions where the job was paused.*Shown as gibibyte* | | **gcp.dataflow.worker.accelerator.gpu.clock.device_memory_current**(gauge) | The current frequency of the memory clock.*Shown as megahertz* | | **gcp.dataflow.worker.accelerator.gpu.clock.device_memory_max**(gauge) | The max frequency of the memory clock.*Shown as megahertz* | | **gcp.dataflow.worker.accelerator.gpu.clock.device_sm_current**(gauge) | The current frequency of the SM clock.*Shown as megahertz* | | **gcp.dataflow.worker.accelerator.gpu.clock.device_sm_max**(gauge) | The max frequency of the SM clock.*Shown as megahertz* | | **gcp.dataflow.worker.accelerator.gpu.memory.device_limit**(gauge) | Total available memory on the device. | | **gcp.dataflow.worker.accelerator.gpu.memory.device_usage**(gauge) | Memory in use on the device, including reserved memory. | | **gcp.dataflow.worker.accelerator.gpu.memory.ecc.volatile.device_correctable_total**(gauge) | The total number of correctable (single-bit) ECC errors on the device. | | **gcp.dataflow.worker.accelerator.gpu.memory.ecc.volatile.device_uncorrectable_total**(gauge) | The total number of uncorrectable (double-bit) ECC errors on the device. | | **gcp.dataflow.worker.accelerator.gpu.nvlink.device_read**(gauge) | The rate that data is read by the device from other GPUs on the host. | | **gcp.dataflow.worker.accelerator.gpu.nvlink.device_transfer**(gauge) | The rate that data is transferred from the device to other GPUs on the host. | | **gcp.dataflow.worker.accelerator.gpu.pcie.device_read**(gauge) | The rate that data is read by the device from the host over PCIe. | | **gcp.dataflow.worker.accelerator.gpu.pcie.device_transfer**(gauge) | The rate that data is transmitted from the device to the host over PCIe. | | **gcp.dataflow.worker.accelerator.gpu.power.device_limit**(gauge) | The power limit of the device.*Shown as watt* | | **gcp.dataflow.worker.accelerator.gpu.power.device_usage**(gauge) | The current power draw of the device.*Shown as watt* | | **gcp.dataflow.worker.accelerator.gpu.temperature.device_current**(gauge) | The current temperature of the device.*Shown as degree celsius* | | **gcp.dataflow.worker.accelerator.gpu.temperature.device_max_op**(gauge) | The max operating temperature of the GPU, above which the GPU will try to cool itself to below this temperature.*Shown as degree celsius* | | **gcp.dataflow.worker.accelerator.gpu.temperature.device_shutdown**(gauge) | The temperature at which the GPU will shut down.*Shown as degree celsius* | | **gcp.dataflow.worker.accelerator.gpu.temperature.device_slowdown**(gauge) | The temperature at which the GPU will start throttling.*Shown as degree celsius* | | **gcp.dataflow.worker.accelerator.gpu.utilization.device_fp16_pipe_activity**(gauge) | Percent of time that the FP16 Pipe was active on the device.*Shown as percent* | | **gcp.dataflow.worker.accelerator.gpu.utilization.device_fp32_pipe_activity**(gauge) | Percent of time that the FP32 Pipe was active on the device.*Shown as percent* | | **gcp.dataflow.worker.accelerator.gpu.utilization.device_fp64_pipe_activity**(gauge) | Percent of time that the FP64 Pipe was active on the device.*Shown as percent* | | **gcp.dataflow.worker.accelerator.gpu.utilization.device_kernel_runtime**(gauge) | Percent of time over the last sample period during which at least one kernel was executing on the GPU.*Shown as percent* | | **gcp.dataflow.worker.accelerator.gpu.utilization.device_memory_access**(gauge) | Percent of time over the last sample period during which global device memory was being read or written.*Shown as percent* | | **gcp.dataflow.worker.accelerator.gpu.utilization.device_sm_activity**(gauge) | Percent of time a warp was active on an SM averaged across all SMs on the device.*Shown as percent* | | **gcp.dataflow.worker.accelerator.gpu.utilization.device_sm_occupancy**(gauge) | Percent of the active warps on the device relative to the max.*Shown as percent* | | **gcp.dataflow.worker.accelerator.gpu.utilization.device_tensor_pipe_activity**(gauge) | Percent of time that the Tensor Pipe was active on the device.*Shown as percent* | {% alert level="warning" %} When using Google Cloud Dataflow to monitor Apache Beam pipeline metrics, note that metrics generated from [Gauge static methods](https://beam.apache.org/releases/javadoc/current/org/apache/beam/sdk/metrics/Metrics.html) are not collected. If you need to monitor these metrics, you can use [Micrometer](https://micrometer.io/docs). {% /alert %} ### Events{% #events %} The Google Cloud Dataflow integration does not include any events. ### Service Checks{% #service-checks %} The Google Cloud Dataflow integration does not include any service checks. ## Troubleshooting{% #troubleshooting %} Need help? Contact [Datadog support](https://docs.datadoghq.com/help/). ## Further Reading{% #further-reading %} Additional helpful documentation, links, and articles: - [Monitor your Dataflow pipelines with Datadog](https://www.datadoghq.com/blog/monitor-dataflow-pipelines-with-datadog/)