prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  Python 3 support in Dataflow component
  Apache beam has had python 3 support since version 2.14.0, and python 2 will reach EOL in January 2020. Right now it looks like using `python2` is hard coded into the [Dataflow component](https://github.com/kubeflow/pipelines/blob/master/components/gcp/container/component_sdk/python/kfp_component/google/dataflow/_launch_python.py):
  
  ```python
  def _prepare_cmd(project_id, python_file_path, args, staging_location):
      dataflow_args = [
          '--runner', 'dataflow', 
          '--project', project_id]
      if staging_location:
          dataflow_args += ['--staging_location', staging_location, '--temp_location', staging_location]
      return (['python2', '-u', python_file_path] + 
          dataflow_args + args)
  ```
  
  Can an option be added to run Dataflow jobs using `python3`?
  Python 3 support in Dataflow component
  Apache beam has had python 3 support since version 2.14.0, and python 2 will reach EOL in January 2020. Right now it looks like using `python2` is hard coded into the [Dataflow component](https://github.com/kubeflow/pipelines/blob/master/components/gcp/container/component_sdk/python/kfp_component/google/dataflow/_launch_python.py):
  
  ```python
  def _prepare_cmd(project_id, python_file_path, args, staging_location):
      dataflow_args = [
          '--runner', 'dataflow', 
          '--project', project_id]
      if staging_location:
          dataflow_args += ['--staging_location', staging_location, '--temp_location', staging_location]
      return (['python2', '-u', python_file_path] + 
          dataflow_args + args)
  ```
  
  Can an option be added to run Dataflow jobs using `python3`?
  
  --- END ISSUE ---
  
  Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
  --- BEGIN FILES ---
  Path: `components/gcp/container/component_sdk/python/kfp_component/google/dataflow/_launch_python.py`
  Content:
  ```
  1 # Copyright 2018 Google LLC
  2 #
  3 # Licensed under the Apache License, Version 2.0 (the "License");
  4 # you may not use this file except in compliance with the License.
  5 # You may obtain a copy of the License at
  6 #
  7 #      http://www.apache.org/licenses/LICENSE-2.0
  8 #
  9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14 import subprocess
  15 import re
  16 import logging
  17 import os
  18 
  19 from google.cloud import storage
  20 from kfp_component.core import KfpExecutionContext
  21 from ._client import DataflowClient
  22 from ._common_ops import (wait_and_dump_job, stage_file, get_staging_location, 
  23     read_job_id_and_location, upload_job_id_and_location)
  24 from ._process import Process
  25 from ..storage import parse_blob_path
  26 
  27 def launch_python(python_file_path, project_id, staging_dir=None, requirements_file_path=None, 
  28     args=[], wait_interval=30,
  29     job_id_output_path='/tmp/kfp/output/dataflow/job_id.txt',
  30     job_object_output_path='/tmp/kfp/output/dataflow/job.json',
  31 ):
  32     """Launch a self-executing beam python file.
  33 
  34     Args:
  35         python_file_path (str): The gcs or local path to the python file to run.
  36         project_id (str): The ID of the parent project.
  37         staging_dir (str): Optional. The GCS directory for keeping staging files. 
  38             A random subdirectory will be created under the directory to keep job info
  39             for resuming the job in case of failure and it will be passed as 
  40             `staging_location` and `temp_location` command line args of the beam code.
  41         requirements_file_path (str): Optional, the gcs or local path to the pip 
  42             requirements file.
  43         args (list): The list of args to pass to the python file.
  44         wait_interval (int): The wait seconds between polling.
  45     Returns:
  46         The completed job.
  47     """
  48     storage_client = storage.Client()
  49     df_client = DataflowClient()
  50     job_id = None
  51     location = None
  52     def cancel():
  53         if job_id:
  54             df_client.cancel_job(
  55                 project_id,
  56                 job_id,
  57                 location
  58             )
  59     with KfpExecutionContext(on_cancel=cancel) as ctx:
  60         staging_location = get_staging_location(staging_dir, ctx.context_id())
  61         job_id, location = read_job_id_and_location(storage_client, staging_location)
  62         # Continue waiting for the job if it's has been uploaded to staging location.
  63         if job_id:
  64             job = df_client.get_job(project_id, job_id, location)
  65             return wait_and_dump_job(df_client, project_id, location, job,
  66                 wait_interval,
  67                 job_id_output_path=job_id_output_path,
  68                 job_object_output_path=job_object_output_path,
  69             )
  70 
  71         _install_requirements(requirements_file_path)
  72         python_file_path = stage_file(python_file_path)
  73         cmd = _prepare_cmd(project_id, python_file_path, args, staging_location)
  74         sub_process = Process(cmd)
  75         for line in sub_process.read_lines():
  76             job_id, location = _extract_job_id_and_location(line)
  77             if job_id:
  78                 logging.info('Found job id {} and location {}.'.format(job_id, location))
  79                 upload_job_id_and_location(storage_client, staging_location, job_id, location)
  80                 break
  81         sub_process.wait_and_check()
  82         if not job_id:
  83             logging.warning('No dataflow job was found when '
  84                 'running the python file.')
  85             return None
  86         job = df_client.get_job(project_id, job_id, 
  87             location=location)
  88         return wait_and_dump_job(df_client, project_id, location, job,
  89             wait_interval,
  90             job_id_output_path=job_id_output_path,
  91             job_object_output_path=job_object_output_path,
  92         )
  93 
  94 def _prepare_cmd(project_id, python_file_path, args, staging_location):
  95     dataflow_args = [
  96         '--runner', 'dataflow', 
  97         '--project', project_id]
  98     if staging_location:
  99         dataflow_args += ['--staging_location', staging_location, '--temp_location', staging_location]
  100     return (['python2', '-u', python_file_path] + 
  101         dataflow_args + args)
  102 
  103 def _extract_job_id_and_location(line):
  104     """Returns (job_id, location) from matched log.
  105     """
  106     job_id_pattern = re.compile(
  107         br'.*console.cloud.google.com/dataflow.*/locations/([a-z|0-9|A-Z|\-|\_]+)/jobs/([a-z|0-9|A-Z|\-|\_]+).*')
  108     matched_job_id = job_id_pattern.search(line or '')
  109     if matched_job_id:
  110         return (matched_job_id.group(2).decode(), matched_job_id.group(1).decode())
  111     return (None, None)
  112 
  113 def _install_requirements(requirements_file_path):
  114     if not requirements_file_path:
  115         return
  116     requirements_file_path = stage_file(requirements_file_path)
  117     subprocess.check_call(['pip2', 'install', '-r', requirements_file_path])
  ```
  --- END FILES ---
  
  Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.
  
  Here is an example:
  
  ```diff
  diff --git a/examples/server_async.py b/examples/server_async.py
  --- a/examples/server_async.py
  +++ b/examples/server_async.py
  @@ -313,4 +313,4 @@
   
   
   if __name__ == "__main__":
  -    asyncio.run(run_async_server("."), debug=True)
  +    asyncio.run(run_async_server(), debug=True)
  diff --git a/examples/server_sync.py b/examples/server_sync.py
  --- a/examples/server_sync.py
  +++ b/examples/server_sync.py
  @@ -313,5 +313,5 @@
   
   
   if __name__ == "__main__":
  -    server = run_sync_server(".")
  +    server = run_sync_server()
       server.shutdown()
  
  ```
  
golden_diff: diff --git a/components/gcp/container/component_sdk/python/kfp_component/google/dataflow/_launch_python.py b/components/gcp/container/component_sdk/python/kfp_component/google/dataflow/_launch_python.py
  --- a/components/gcp/container/component_sdk/python/kfp_component/google/dataflow/_launch_python.py
  +++ b/components/gcp/container/component_sdk/python/kfp_component/google/dataflow/_launch_python.py
  @@ -97,7 +97,7 @@
           '--project', project_id]
       if staging_location:
           dataflow_args += ['--staging_location', staging_location, '--temp_location', staging_location]
  -    return (['python2', '-u', python_file_path] + 
  +    return (['python', '-u', python_file_path] + 
           dataflow_args + args)
   
   def _extract_job_id_and_location(line):
  @@ -114,4 +114,4 @@
       if not requirements_file_path:
           return
       requirements_file_path = stage_file(requirements_file_path)
  -    subprocess.check_call(['pip2', 'install', '-r', requirements_file_path])
  \ No newline at end of file
  +    subprocess.check_call(['pip', 'install', '-r', requirements_file_path])
  \ No newline at end of file
  
verification_info: {"golden_diff": "diff --git a/components/gcp/container/component_sdk/python/kfp_component/google/dataflow/_launch_python.py b/components/gcp/container/component_sdk/python/kfp_component/google/dataflow/_launch_python.py\n--- a/components/gcp/container/component_sdk/python/kfp_component/google/dataflow/_launch_python.py\n+++ b/components/gcp/container/component_sdk/python/kfp_component/google/dataflow/_launch_python.py\n@@ -97,7 +97,7 @@\n         '--project', project_id]\n     if staging_location:\n         dataflow_args += ['--staging_location', staging_location, '--temp_location', staging_location]\n-    return (['python2', '-u', python_file_path] + \n+    return (['python', '-u', python_file_path] + \n         dataflow_args + args)\n \n def _extract_job_id_and_location(line):\n@@ -114,4 +114,4 @@\n     if not requirements_file_path:\n         return\n     requirements_file_path = stage_file(requirements_file_path)\n-    subprocess.check_call(['pip2', 'install', '-r', requirements_file_path])\n\\ No newline at end of file\n+    subprocess.check_call(['pip', 'install', '-r', requirements_file_path])\n\\ No newline at end of file\n", "issue": "Python 3 support in Dataflow component\nApache beam has had python 3 support since version 2.14.0, and python 2 will reach EOL in January 2020. Right now it looks like using `python2` is hard coded into the [Dataflow component](https://github.com/kubeflow/pipelines/blob/master/components/gcp/container/component_sdk/python/kfp_component/google/dataflow/_launch_python.py):\r\n\r\n```python\r\ndef _prepare_cmd(project_id, python_file_path, args, staging_location):\r\n    dataflow_args = [\r\n        '--runner', 'dataflow', \r\n        '--project', project_id]\r\n    if staging_location:\r\n        dataflow_args += ['--staging_location', staging_location, '--temp_location', staging_location]\r\n    return (['python2', '-u', python_file_path] + \r\n        dataflow_args + args)\r\n```\r\n\r\nCan an option be added to run Dataflow jobs using `python3`?\nPython 3 support in Dataflow component\nApache beam has had python 3 support since version 2.14.0, and python 2 will reach EOL in January 2020. Right now it looks like using `python2` is hard coded into the [Dataflow component](https://github.com/kubeflow/pipelines/blob/master/components/gcp/container/component_sdk/python/kfp_component/google/dataflow/_launch_python.py):\r\n\r\n```python\r\ndef _prepare_cmd(project_id, python_file_path, args, staging_location):\r\n    dataflow_args = [\r\n        '--runner', 'dataflow', \r\n        '--project', project_id]\r\n    if staging_location:\r\n        dataflow_args += ['--staging_location', staging_location, '--temp_location', staging_location]\r\n    return (['python2', '-u', python_file_path] + \r\n        dataflow_args + args)\r\n```\r\n\r\nCan an option be added to run Dataflow jobs using `python3`?\n", "before_files": [{"content": "# Copyright 2018 Google LLC\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\nimport subprocess\nimport re\nimport logging\nimport os\n\nfrom google.cloud import storage\nfrom kfp_component.core import KfpExecutionContext\nfrom ._client import DataflowClient\nfrom ._common_ops import (wait_and_dump_job, stage_file, get_staging_location, \n    read_job_id_and_location, upload_job_id_and_location)\nfrom ._process import Process\nfrom ..storage import parse_blob_path\n\ndef launch_python(python_file_path, project_id, staging_dir=None, requirements_file_path=None, \n    args=[], wait_interval=30,\n    job_id_output_path='/tmp/kfp/output/dataflow/job_id.txt',\n    job_object_output_path='/tmp/kfp/output/dataflow/job.json',\n):\n    \"\"\"Launch a self-executing beam python file.\n\n    Args:\n        python_file_path (str): The gcs or local path to the python file to run.\n        project_id (str): The ID of the parent project.\n        staging_dir (str): Optional. The GCS directory for keeping staging files. \n            A random subdirectory will be created under the directory to keep job info\n            for resuming the job in case of failure and it will be passed as \n            `staging_location` and `temp_location` command line args of the beam code.\n        requirements_file_path (str): Optional, the gcs or local path to the pip \n            requirements file.\n        args (list): The list of args to pass to the python file.\n        wait_interval (int): The wait seconds between polling.\n    Returns:\n        The completed job.\n    \"\"\"\n    storage_client = storage.Client()\n    df_client = DataflowClient()\n    job_id = None\n    location = None\n    def cancel():\n        if job_id:\n            df_client.cancel_job(\n                project_id,\n                job_id,\n                location\n            )\n    with KfpExecutionContext(on_cancel=cancel) as ctx:\n        staging_location = get_staging_location(staging_dir, ctx.context_id())\n        job_id, location = read_job_id_and_location(storage_client, staging_location)\n        # Continue waiting for the job if it's has been uploaded to staging location.\n        if job_id:\n            job = df_client.get_job(project_id, job_id, location)\n            return wait_and_dump_job(df_client, project_id, location, job,\n                wait_interval,\n                job_id_output_path=job_id_output_path,\n                job_object_output_path=job_object_output_path,\n            )\n\n        _install_requirements(requirements_file_path)\n        python_file_path = stage_file(python_file_path)\n        cmd = _prepare_cmd(project_id, python_file_path, args, staging_location)\n        sub_process = Process(cmd)\n        for line in sub_process.read_lines():\n            job_id, location = _extract_job_id_and_location(line)\n            if job_id:\n                logging.info('Found job id {} and location {}.'.format(job_id, location))\n                upload_job_id_and_location(storage_client, staging_location, job_id, location)\n                break\n        sub_process.wait_and_check()\n        if not job_id:\n            logging.warning('No dataflow job was found when '\n                'running the python file.')\n            return None\n        job = df_client.get_job(project_id, job_id, \n            location=location)\n        return wait_and_dump_job(df_client, project_id, location, job,\n            wait_interval,\n            job_id_output_path=job_id_output_path,\n            job_object_output_path=job_object_output_path,\n        )\n\ndef _prepare_cmd(project_id, python_file_path, args, staging_location):\n    dataflow_args = [\n        '--runner', 'dataflow', \n        '--project', project_id]\n    if staging_location:\n        dataflow_args += ['--staging_location', staging_location, '--temp_location', staging_location]\n    return (['python2', '-u', python_file_path] + \n        dataflow_args + args)\n\ndef _extract_job_id_and_location(line):\n    \"\"\"Returns (job_id, location) from matched log.\n    \"\"\"\n    job_id_pattern = re.compile(\n        br'.*console.cloud.google.com/dataflow.*/locations/([a-z|0-9|A-Z|\\-|\\_]+)/jobs/([a-z|0-9|A-Z|\\-|\\_]+).*')\n    matched_job_id = job_id_pattern.search(line or '')\n    if matched_job_id:\n        return (matched_job_id.group(2).decode(), matched_job_id.group(1).decode())\n    return (None, None)\n\ndef _install_requirements(requirements_file_path):\n    if not requirements_file_path:\n        return\n    requirements_file_path = stage_file(requirements_file_path)\n    subprocess.check_call(['pip2', 'install', '-r', requirements_file_path])", "path": "components/gcp/container/component_sdk/python/kfp_component/google/dataflow/_launch_python.py"}], "after_files": [{"content": "# Copyright 2018 Google LLC\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\nimport subprocess\nimport re\nimport logging\nimport os\n\nfrom google.cloud import storage\nfrom kfp_component.core import KfpExecutionContext\nfrom ._client import DataflowClient\nfrom ._common_ops import (wait_and_dump_job, stage_file, get_staging_location, \n    read_job_id_and_location, upload_job_id_and_location)\nfrom ._process import Process\nfrom ..storage import parse_blob_path\n\ndef launch_python(python_file_path, project_id, staging_dir=None, requirements_file_path=None, \n    args=[], wait_interval=30,\n    job_id_output_path='/tmp/kfp/output/dataflow/job_id.txt',\n    job_object_output_path='/tmp/kfp/output/dataflow/job.json',\n):\n    \"\"\"Launch a self-executing beam python file.\n\n    Args:\n        python_file_path (str): The gcs or local path to the python file to run.\n        project_id (str): The ID of the parent project.\n        staging_dir (str): Optional. The GCS directory for keeping staging files. \n            A random subdirectory will be created under the directory to keep job info\n            for resuming the job in case of failure and it will be passed as \n            `staging_location` and `temp_location` command line args of the beam code.\n        requirements_file_path (str): Optional, the gcs or local path to the pip \n            requirements file.\n        args (list): The list of args to pass to the python file.\n        wait_interval (int): The wait seconds between polling.\n    Returns:\n        The completed job.\n    \"\"\"\n    storage_client = storage.Client()\n    df_client = DataflowClient()\n    job_id = None\n    location = None\n    def cancel():\n        if job_id:\n            df_client.cancel_job(\n                project_id,\n                job_id,\n                location\n            )\n    with KfpExecutionContext(on_cancel=cancel) as ctx:\n        staging_location = get_staging_location(staging_dir, ctx.context_id())\n        job_id, location = read_job_id_and_location(storage_client, staging_location)\n        # Continue waiting for the job if it's has been uploaded to staging location.\n        if job_id:\n            job = df_client.get_job(project_id, job_id, location)\n            return wait_and_dump_job(df_client, project_id, location, job,\n                wait_interval,\n                job_id_output_path=job_id_output_path,\n                job_object_output_path=job_object_output_path,\n            )\n\n        _install_requirements(requirements_file_path)\n        python_file_path = stage_file(python_file_path)\n        cmd = _prepare_cmd(project_id, python_file_path, args, staging_location)\n        sub_process = Process(cmd)\n        for line in sub_process.read_lines():\n            job_id, location = _extract_job_id_and_location(line)\n            if job_id:\n                logging.info('Found job id {} and location {}.'.format(job_id, location))\n                upload_job_id_and_location(storage_client, staging_location, job_id, location)\n                break\n        sub_process.wait_and_check()\n        if not job_id:\n            logging.warning('No dataflow job was found when '\n                'running the python file.')\n            return None\n        job = df_client.get_job(project_id, job_id, \n            location=location)\n        return wait_and_dump_job(df_client, project_id, location, job,\n            wait_interval,\n            job_id_output_path=job_id_output_path,\n            job_object_output_path=job_object_output_path,\n        )\n\ndef _prepare_cmd(project_id, python_file_path, args, staging_location):\n    dataflow_args = [\n        '--runner', 'dataflow', \n        '--project', project_id]\n    if staging_location:\n        dataflow_args += ['--staging_location', staging_location, '--temp_location', staging_location]\n    return (['python', '-u', python_file_path] + \n        dataflow_args + args)\n\ndef _extract_job_id_and_location(line):\n    \"\"\"Returns (job_id, location) from matched log.\n    \"\"\"\n    job_id_pattern = re.compile(\n        br'.*console.cloud.google.com/dataflow.*/locations/([a-z|0-9|A-Z|\\-|\\_]+)/jobs/([a-z|0-9|A-Z|\\-|\\_]+).*')\n    matched_job_id = job_id_pattern.search(line or '')\n    if matched_job_id:\n        return (matched_job_id.group(2).decode(), matched_job_id.group(1).decode())\n    return (None, None)\n\ndef _install_requirements(requirements_file_path):\n    if not requirements_file_path:\n        return\n    requirements_file_path = stage_file(requirements_file_path)\n    subprocess.check_call(['pip', 'install', '-r', requirements_file_path])", "path": "components/gcp/container/component_sdk/python/kfp_component/google/dataflow/_launch_python.py"}]}