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()

```

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

{"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"}]}