You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by bc...@apache.org on 2016/12/07 01:15:44 UTC
[2/4] incubator-beam git commit: [BEAM-1047] Add DataflowClient
wrapper on top of JSON library.
[BEAM-1047] Add DataflowClient wrapper on top of JSON library.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/e8c9686a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/e8c9686a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/e8c9686a
Branch: refs/heads/master
Commit: e8c9686a2e898d38afd692328eb171c542084747
Parents: ded5883
Author: Pei He <pe...@google.com>
Authored: Wed Nov 23 15:59:56 2016 -0800
Committer: bchambers <bc...@google.com>
Committed: Tue Dec 6 17:08:12 2016 -0800
----------------------------------------------------------------------
.../beam/runners/dataflow/DataflowClient.java | 130 +++++++++++++++++++
1 file changed, 130 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/e8c9686a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowClient.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowClient.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowClient.java
new file mode 100644
index 0000000..f2081db
--- /dev/null
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowClient.java
@@ -0,0 +1,130 @@
+/*
+ * 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.runners.dataflow;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.api.services.dataflow.Dataflow;
+import com.google.api.services.dataflow.Dataflow.Projects.Jobs;
+import com.google.api.services.dataflow.model.Job;
+import com.google.api.services.dataflow.model.JobMetrics;
+import com.google.api.services.dataflow.model.LeaseWorkItemRequest;
+import com.google.api.services.dataflow.model.LeaseWorkItemResponse;
+import com.google.api.services.dataflow.model.ListJobMessagesResponse;
+import com.google.api.services.dataflow.model.ListJobsResponse;
+import com.google.api.services.dataflow.model.ReportWorkItemStatusRequest;
+import com.google.api.services.dataflow.model.ReportWorkItemStatusResponse;
+import java.io.IOException;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
+
+/**
+ * Client library for {@link Dataflow}.
+ */
+public class DataflowClient {
+
+ public static DataflowClient create(DataflowPipelineOptions options) {
+ return new DataflowClient(options.getDataflowClient(), options);
+ }
+
+ private final Dataflow dataflow;
+ private final DataflowPipelineOptions options;
+
+ private DataflowClient(Dataflow dataflow, DataflowPipelineOptions options) {
+ this.dataflow = checkNotNull(dataflow, "dataflow");
+ this.options = checkNotNull(options, "options");
+ }
+
+ /**
+ * Creates the Dataflow {@link Job}.
+ */
+ public Job createJob(@Nonnull Job job) throws IOException {
+ Jobs.Create jobsCreate = dataflow.projects().jobs().create(options.getProject(), job);
+ return jobsCreate.execute();
+ }
+
+ /**
+ * Lists Dataflow {@link Job Jobs} in the project associated with
+ * the {@link DataflowPipelineOptions}.
+ */
+ public ListJobsResponse listJobs(@Nullable String pageToken) throws IOException {
+ Jobs.List jobsList = dataflow.projects().jobs()
+ .list(options.getProject())
+ .setPageToken(pageToken);
+ return jobsList.execute();
+ }
+
+ /**
+ * Updates the Dataflow {@link Job} with the given {@code jobId}.
+ */
+ public Job updateJob(@Nonnull String jobId, @Nonnull Job content) throws IOException {
+ Jobs.Update jobsUpdate = dataflow.projects().jobs()
+ .update(options.getProject(), jobId, content);
+ return jobsUpdate.execute();
+ }
+
+ /**
+ * Gets the Dataflow {@link Job} with the given {@code jobId}.
+ */
+ public Job getJob(@Nonnull String jobId) throws IOException {
+ Jobs.Get jobsGet = dataflow.projects().jobs()
+ .get(options.getProject(), jobId);
+ return jobsGet.execute();
+ }
+
+ /**
+ * Gets the {@link JobMetrics} with the given {@code jobId}.
+ */
+ public JobMetrics getJobMetrics(@Nonnull String jobId) throws IOException {
+ Jobs.GetMetrics jobsGetMetrics = dataflow.projects().jobs()
+ .getMetrics(options.getProject(), jobId);
+ return jobsGetMetrics.execute();
+ }
+
+ /**
+ * Lists job messages with the given {@code jobId}.
+ */
+ public ListJobMessagesResponse listJobMessages(
+ @Nonnull String jobId, @Nullable String pageToken) throws IOException {
+ Jobs.Messages.List jobMessagesList = dataflow.projects().jobs().messages()
+ .list(options.getProject(), jobId)
+ .setPageToken(pageToken);
+ return jobMessagesList.execute();
+ }
+
+ /**
+ * Leases the work item for {@code jobId}.
+ */
+ public LeaseWorkItemResponse leaseWorkItem(
+ @Nonnull String jobId, @Nonnull LeaseWorkItemRequest request) throws IOException {
+ Jobs.WorkItems.Lease jobWorkItemsLease = dataflow.projects().jobs().workItems()
+ .lease(options.getProject(), jobId, request);
+ return jobWorkItemsLease.execute();
+ }
+
+ /**
+ * Reports the status of the work item for {@code jobId}.
+ */
+ public ReportWorkItemStatusResponse reportWorkItemStatus(
+ @Nonnull String jobId, @Nonnull ReportWorkItemStatusRequest request) throws IOException {
+ Jobs.WorkItems.ReportStatus jobWorkItemsReportStatus = dataflow.projects().jobs().workItems()
+ .reportStatus(options.getProject(), jobId, request);
+ return jobWorkItemsReportStatus.execute();
+ }
+}