| # |
| # 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. |
| # |
| |
| """A job server submitting portable pipelines as uber jars to Flink.""" |
| |
| # pytype: skip-file |
| |
| from __future__ import absolute_import |
| from __future__ import print_function |
| |
| import logging |
| import os |
| import tempfile |
| import time |
| import urllib |
| |
| import requests |
| from google.protobuf import json_format |
| |
| from apache_beam.options import pipeline_options |
| from apache_beam.portability.api import beam_job_api_pb2 |
| from apache_beam.runners.portability import abstract_job_service |
| from apache_beam.runners.portability import job_server |
| |
| _LOGGER = logging.getLogger(__name__) |
| |
| |
| class FlinkUberJarJobServer(abstract_job_service.AbstractJobServiceServicer): |
| """A Job server which submits a self-contained Jar to a Flink cluster. |
| |
| The jar contains the Beam pipeline definition, dependencies, and |
| the pipeline artifacts. |
| """ |
| def __init__(self, master_url, options): |
| super(FlinkUberJarJobServer, self).__init__() |
| self._master_url = master_url |
| self._executable_jar = ( |
| options.view_as( |
| pipeline_options.FlinkRunnerOptions).flink_job_server_jar) |
| self._artifact_port = ( |
| options.view_as(pipeline_options.JobServerOptions).artifact_port) |
| self._temp_dir = tempfile.mkdtemp(prefix='apache-beam-flink') |
| |
| def start(self): |
| return self |
| |
| def stop(self): |
| pass |
| |
| def executable_jar(self): |
| if self._executable_jar: |
| if not os.path.exists(self._executable_jar): |
| parsed = urllib.parse.urlparse(self._executable_jar) |
| if not parsed.scheme: |
| raise ValueError( |
| 'Unable to parse jar URL "%s". If using a full URL, make sure ' |
| 'the scheme is specified. If using a local file path, make sure ' |
| 'the file exists; you may have to first build the job server ' |
| 'using `./gradlew runners:flink:%s:job-server:shadowJar`.' % |
| (self._executable_jar, self._flink_version)) |
| url = self._executable_jar |
| else: |
| url = job_server.JavaJarJobServer.path_to_beam_jar( |
| 'runners:flink:%s:job-server:shadowJar' % self.flink_version()) |
| return job_server.JavaJarJobServer.local_jar(url) |
| |
| def flink_version(self): |
| full_version = requests.get('%s/v1/config' % |
| self._master_url).json()['flink-version'] |
| # Only return up to minor version. |
| return '.'.join(full_version.split('.')[:2]) |
| |
| def create_beam_job(self, job_id, job_name, pipeline, options): |
| return FlinkBeamJob( |
| self._master_url, |
| self.executable_jar(), |
| job_id, |
| job_name, |
| pipeline, |
| options, |
| artifact_port=self._artifact_port) |
| |
| def GetJobMetrics(self, request, context=None): |
| if request.job_id not in self._jobs: |
| raise LookupError("Job {} does not exist".format(request.job_id)) |
| metrics_text = self._jobs[request.job_id].get_metrics() |
| response = beam_job_api_pb2.GetJobMetricsResponse() |
| json_format.Parse(metrics_text, response) |
| return response |
| |
| |
| class FlinkBeamJob(abstract_job_service.UberJarBeamJob): |
| """Runs a single Beam job on Flink by staging all contents into a Jar |
| and uploading it via the Flink Rest API.""" |
| def __init__( |
| self, |
| master_url, |
| executable_jar, |
| job_id, |
| job_name, |
| pipeline, |
| options, |
| artifact_port=0): |
| super(FlinkBeamJob, self).__init__( |
| executable_jar, |
| job_id, |
| job_name, |
| pipeline, |
| options, |
| artifact_port=artifact_port) |
| self._master_url = master_url |
| |
| def request(self, method, path, expected_status=200, **kwargs): |
| url = '%s/%s' % (self._master_url, path) |
| response = method(url, **kwargs) |
| if response.status_code != expected_status: |
| raise RuntimeError( |
| "Request to %s failed with status %d: %s" % |
| (url, response.status_code, response.text)) |
| if response.text: |
| return response.json() |
| |
| def get(self, path, **kwargs): |
| return self.request(requests.get, path, **kwargs) |
| |
| def post(self, path, **kwargs): |
| return self.request(requests.post, path, **kwargs) |
| |
| def delete(self, path, **kwargs): |
| return self.request(requests.delete, path, **kwargs) |
| |
| def run(self): |
| self._stop_artifact_service() |
| |
| # Upload the jar and start the job. |
| with open(self._jar, 'rb') as jar_file: |
| self._flink_jar_id = self.post( |
| 'v1/jars/upload', |
| files={'jarfile': ('beam.jar', jar_file)})['filename'].split('/')[-1] |
| self._jar_uploaded = True |
| self._flink_job_id = self.post( |
| 'v1/jars/%s/run' % self._flink_jar_id, |
| json={ |
| 'entryClass': 'org.apache.beam.runners.flink.FlinkPipelineRunner' |
| })['jobid'] |
| os.unlink(self._jar) |
| _LOGGER.info('Started Flink job as %s' % self._flink_job_id) |
| |
| def cancel(self): |
| self.post('v1/%s/stop' % self._flink_job_id, expected_status=202) |
| self.delete_jar() |
| |
| def delete_jar(self): |
| if self._jar_uploaded: |
| self._jar_uploaded = False |
| try: |
| self.delete('v1/jars/%s' % self._flink_jar_id) |
| except Exception: |
| _LOGGER.info( |
| 'Error deleting jar %s' % self._flink_jar_id, exc_info=True) |
| |
| def _get_state(self): |
| """Query flink to get the current state. |
| |
| :return: tuple of int and Timestamp or None |
| timestamp will be None if the state has not changed since the last query. |
| """ |
| # For just getting the status, execution-result seems cheaper. |
| flink_status = self.get('v1/jobs/%s/execution-result' % |
| self._flink_job_id)['status']['id'] |
| if flink_status == 'COMPLETED': |
| flink_status = self.get('v1/jobs/%s' % self._flink_job_id)['state'] |
| beam_state = { |
| 'CREATED': beam_job_api_pb2.JobState.STARTING, |
| 'RUNNING': beam_job_api_pb2.JobState.RUNNING, |
| 'FAILING': beam_job_api_pb2.JobState.RUNNING, |
| 'FAILED': beam_job_api_pb2.JobState.FAILED, |
| 'CANCELLING': beam_job_api_pb2.JobState.CANCELLING, |
| 'CANCELED': beam_job_api_pb2.JobState.CANCELLED, |
| 'FINISHED': beam_job_api_pb2.JobState.DONE, |
| 'RESTARTING': beam_job_api_pb2.JobState.RUNNING, |
| 'SUSPENDED': beam_job_api_pb2.JobState.RUNNING, |
| 'RECONCILING': beam_job_api_pb2.JobState.RUNNING, |
| 'IN_PROGRESS': beam_job_api_pb2.JobState.RUNNING, |
| 'COMPLETED': beam_job_api_pb2.JobState.DONE, |
| }.get(flink_status, beam_job_api_pb2.JobState.UNSPECIFIED) |
| if self.is_terminal_state(beam_state): |
| self.delete_jar() |
| # update the state history if it has changed |
| return beam_state, self.set_state(beam_state) |
| |
| def get_state(self): |
| state, timestamp = self._get_state() |
| if timestamp is None: |
| # state has not changed since it was last checked: use previous timestamp |
| return super(FlinkBeamJob, self).get_state() |
| else: |
| return state, timestamp |
| |
| def get_state_stream(self): |
| def _state_iter(): |
| sleep_secs = 1.0 |
| while True: |
| yield self.get_state() |
| sleep_secs = min(60, sleep_secs * 1.2) |
| time.sleep(sleep_secs) |
| |
| for state, timestamp in self.with_state_history(_state_iter()): |
| yield state, timestamp |
| if self.is_terminal_state(state): |
| break |
| |
| def get_message_stream(self): |
| for state, timestamp in self.get_state_stream(): |
| if self.is_terminal_state(state): |
| response = self.get('v1/jobs/%s/exceptions' % self._flink_job_id) |
| for ix, exc in enumerate(response['all-exceptions']): |
| yield beam_job_api_pb2.JobMessage( |
| message_id='message%d' % ix, |
| time=str(exc['timestamp']), |
| importance=beam_job_api_pb2.JobMessage.MessageImportance. |
| JOB_MESSAGE_ERROR, |
| message_text=exc['exception']) |
| yield state, timestamp |
| break |
| else: |
| yield state, timestamp |
| |
| def get_metrics(self): |
| accumulators = self.get('v1/jobs/%s/accumulators' % |
| self._flink_job_id)['user-task-accumulators'] |
| for accumulator in accumulators: |
| if accumulator['name'] == '__metricscontainers': |
| return accumulator['value'] |
| raise LookupError( |
| "Found no metrics container for job {}".format(self._flink_job_id)) |
