[
https://issues.apache.org/jira/browse/BEAM-10317?focusedWorklogId=453744&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-453744
]
ASF GitHub Bot logged work on BEAM-10317:
-----------------------------------------
Author: ASF GitHub Bot
Created on: 02/Jul/20 02:10
Start Date: 02/Jul/20 02:10
Worklog Time Spent: 10m
Work Description: ajamato commented on a change in pull request #12083:
URL: https://github.com/apache/beam/pull/12083#discussion_r448693178
##########
File path:
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImpl.java
##########
@@ -216,11 +221,11 @@ public void startQueryJob(JobReference jobRef,
JobConfigurationQuery queryConfig
@Override
public void startCopyJob(JobReference jobRef, JobConfigurationTableCopy
copyConfig)
throws IOException, InterruptedException {
- Job job =
- new Job()
- .setJobReference(jobRef)
- .setConfiguration(new JobConfiguration().setCopy(copyConfig));
-
+ Map<String, String> labelMap = new HashMap<>();
+ JobConfiguration config = new JobConfiguration();
+ config.setCopy(copyConfig);
+ config.setLabels(this.bqIOMetadata.addAdditionalJobLabels(labelMap));
+ Job job = new Job().setJobReference(jobRef).setConfiguration(config);
Review comment:
Done
##########
File path:
sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GceMetadataUtil.java
##########
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.extensions.gcp.util;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.Reader;
+import java.nio.charset.Charset;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.io.CharStreams;
+import org.apache.http.HttpResponse;
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.impl.client.DefaultHttpClient;
+import org.apache.http.params.BasicHttpParams;
+import org.apache.http.params.HttpConnectionParams;
+import org.apache.http.params.HttpParams;
+
+/** */
+public class GceMetadataUtil {
+ private static final String BASE_METADATA_URL =
"http://metadata/computeMetadata/v1/";;
+
+ static String fetchMetadata(String key) {
+ int timeoutMillis = 5000;
+ final HttpParams httpParams = new BasicHttpParams();
+ HttpConnectionParams.setConnectionTimeout(httpParams, timeoutMillis);
+ HttpClient client = new DefaultHttpClient(httpParams);
+ HttpGet request = new HttpGet(BASE_METADATA_URL + key);
+ request.setHeader("Metadata-Flavor", "Google");
+
+ try {
+ HttpResponse response = client.execute(request);
+ if (response.getStatusLine().getStatusCode() != 200) {
+ // May mean its running on a non DataflowRunner, in which case it's
perfectly normal.
+ return "";
+ }
+ InputStream in = response.getEntity().getContent();
+ try (final Reader reader = new InputStreamReader(in,
Charset.defaultCharset())) {
+ return CharStreams.toString(reader);
Review comment:
If a valid cloud label string is provided job_id is defined in the
metadata, it will be pulled in and added a s a beam_job_id label to the BQ
jobs. Shouldn't break anything, and if the job_id is meaningful it could be
quite useful. It would actually be a good thing to do intentionally, IMO. Might
be worth suggesting to dataproc.
##########
File path:
sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GceMetadataUtil.java
##########
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.extensions.gcp.util;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.Reader;
+import java.nio.charset.Charset;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.io.CharStreams;
+import org.apache.http.HttpResponse;
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.impl.client.DefaultHttpClient;
+import org.apache.http.params.BasicHttpParams;
+import org.apache.http.params.HttpConnectionParams;
+import org.apache.http.params.HttpParams;
+
+/** */
+public class GceMetadataUtil {
+ private static final String BASE_METADATA_URL =
"http://metadata/computeMetadata/v1/";;
+
+ static String fetchMetadata(String key) {
+ int timeoutMillis = 5000;
+ final HttpParams httpParams = new BasicHttpParams();
+ HttpConnectionParams.setConnectionTimeout(httpParams, timeoutMillis);
+ HttpClient client = new DefaultHttpClient(httpParams);
+ HttpGet request = new HttpGet(BASE_METADATA_URL + key);
+ request.setHeader("Metadata-Flavor", "Google");
+
+ try {
+ HttpResponse response = client.execute(request);
+ if (response.getStatusLine().getStatusCode() != 200) {
+ // May mean its running on a non DataflowRunner, in which case it's
perfectly normal.
+ return "";
+ }
+ InputStream in = response.getEntity().getContent();
+ try (final Reader reader = new InputStreamReader(in,
Charset.defaultCharset())) {
+ return CharStreams.toString(reader);
+ }
+ } catch (IOException e) {
+ // May mean its running on a non DataflowRunner, in which case it's
perfectly normal.
Review comment:
As discussed on the python PR. It shouldn't resolve the address and
return much faster than the timeout.
If it does resolve for some reason (as you suggested running on a GCE
VM+setting the metadata, but not via Dataflow), I still validate the returned
job_id string is a valid cloud label before attaching it as a bq job label.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 453744)
Time Spent: 4h (was: 3h 50m)
> Update BigQueryIO to tag BigQuery Jobs with the Dataflow Job ID
> ---------------------------------------------------------------
>
> Key: BEAM-10317
> URL: https://issues.apache.org/jira/browse/BEAM-10317
> Project: Beam
> Issue Type: New Feature
> Components: io-java-gcp
> Reporter: Alex Amato
> Assignee: Alex Amato
> Priority: P2
> Time Spent: 4h
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
