You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2021/04/05 18:38:15 UTC
[GitHub] [beam] davidcavazos opened a new pull request #14431: [BEAM-10937] Tour of Beam use FileSystems for I/O
davidcavazos opened a new pull request #14431:
URL: https://github.com/apache/beam/pull/14431
Use Beam FileSystems for I/O to support multiple file systems instead of plain `open`.
The diffs are mostly:
* Change code snippets to use `FileSystems.open` and `FileSystems.create` along with converting to/from bytes.
* Add a small description of why using them instead of just `open`.
R: @aaltay
Staged: https://colab.research.google.com/github/davidcavazos/beam/blob/tour-of-beam-reading-filesystems/examples/notebooks/tour-of-beam/reading-and-writing-data.ipynb
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier).
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
<table>
<thead>
<tr>
<th>Lang</th>
<th>SDK</th>
<th>ULR</th>
<th>Dataflow</th>
<th>Flink</th>
<th>Samza</th>
<th>Spark</th>
<th>Twister2</th>
</tr>
</thead>
<tbody>
<tr>
<td>Go</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/">
<img src="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
<tr>
<td>Java</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_ULR/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_ULR/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Streaming/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2_Streaming/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/badge/icon">
</a>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Cron/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Java11_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Java11_Cron/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_Examples_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_Examples_Dataflow_V2/lastCompletedBuild/badge/icon">
</a><br>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/badge/icon">
</a>
</td>
</tr>
<tr>
<td>Python</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
<tr>
<td>XLang</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
</tbody>
</table>
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
<table>
<thead>
<tr>
<th>---</th>
<th>Java</th>
<th>Python</th>
<th>Go</th>
<th>Website</th>
<th>Whitespace</th>
<th>Typescript</th>
</tr>
</thead>
<tbody>
<tr>
<td>Non-portable</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/badge/icon">
</a><br>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/lastCompletedBuild/">
<img src="https://camo.githubusercontent.com/4565d7b7e907114e6c1d12323408bd903aa252fefed5eeab93701b05c9628a84/68747470733a2f2f63692d6265616d2e6170616368652e6f72672f6a6f622f6265616d5f507265436f6d6d69745f507974686f6e446f636b65725f43726f6e2f62616467652f69636f6e" alt="Build Status" data-canonical-src="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/lastCompletedBuild/">
<img src="https://camo.githubusercontent.com/21afb097a5745440598bee5c59a027b140585eec871c3f3b883200610fabf722/68747470733a2f2f63692d6265616d2e6170616368652e6f72672f6a6f622f6265616d5f507265436f6d6d69745f507974686f6e446f63735f43726f6e2f62616467652f69636f6e" alt="Build Status" data-canonical-src="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
</tr>
<tr>
<td>Portable</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
</tbody>
</table>
See [.test-infra/jenkins/README](https://github.com/apache/beam/blob/master/.test-infra/jenkins/README.md) for trigger phrase, status and link of all Jenkins jobs.
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
--
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:
users@infra.apache.org
[GitHub] [beam] udim commented on a change in pull request #14431: [BEAM-10937] Tour of Beam use FileSystems for I/O
Posted by GitBox <gi...@apache.org>.
udim commented on a change in pull request #14431:
URL: https://github.com/apache/beam/pull/14431#discussion_r618838488
##########
File path: examples/notebooks/tour-of-beam/reading-and-writing-data.ipynb
##########
@@ -348,36 +396,37 @@
"base_uri": "https://localhost:8080/"
},
"id": "m8iXqE1CRnn5",
- "outputId": "5c9fa281-1a05-4d20-ccdf-7899df6cc083"
+ "outputId": "4f28f0f9-6ba9-4814-ddab-7fed39e64640"
},
"source": [
"import apache_beam as beam\n",
- "\n",
- "class Count(beam.PTransform):\n",
- " def __init__(self, n):\n",
- " self.n = n\n",
- "\n",
- " @staticmethod\n",
- " def count(n):\n",
+ "from apache_beam.options.pipeline_options import PipelineOptions\n",
+ "from typing import Iterable\n",
+ "\n",
+ "@beam.ptransform_fn\n",
+ "@beam.typehints.with_input_types(beam.pvalue.PBegin)\n",
+ "@beam.typehints.with_output_types(int)\n",
+ "def Count(pcollection: beam.PCollection, n: int):\n",
Review comment:
```
"def Count(pcollection: beam.pvalue.PBegin, n: int) -> beam.PCollection[int]:\n",
```
--
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:
users@infra.apache.org
[GitHub] [beam] davidcavazos commented on pull request #14431: [BEAM-10937] Tour of Beam use FileSystems for I/O
Posted by GitBox <gi...@apache.org>.
davidcavazos commented on pull request #14431:
URL: https://github.com/apache/beam/pull/14431#issuecomment-821595520
@udim friendly ping on this.
--
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:
users@infra.apache.org
[GitHub] [beam] udim commented on a change in pull request #14431: [BEAM-10937] Tour of Beam use FileSystems for I/O
Posted by GitBox <gi...@apache.org>.
udim commented on a change in pull request #14431:
URL: https://github.com/apache/beam/pull/14431#discussion_r616961819
##########
File path: examples/notebooks/tour-of-beam/reading-and-writing-data.ipynb
##########
@@ -423,28 +422,36 @@
"base_uri": "https://localhost:8080/"
},
"id": "ywVbJxegaZbo",
- "outputId": "5e0adfa3-e685-4fe0-b6b7-bfa3d8469da1"
+ "outputId": "042ee6be-d4d6-4983-9a7c-5f876ba0f655"
},
"source": [
"import apache_beam as beam\n",
+ "from apache_beam.io.filesystems import FileSystems as beam_fs\n",
+ "import codecs\n",
"import csv\n",
- "import glob\n",
"\n",
"class ReadCsvFiles(beam.PTransform):\n",
" def __init__(self, file_patterns):\n",
" self.file_patterns = file_patterns\n",
"\n",
" @staticmethod\n",
+ " def expand_pattern(pattern):\n",
+ " for match_result in beam_fs.match([pattern])[0].metadata_list:\n",
+ " yield match_result.path\n",
+ "\n",
+ " @staticmethod\n",
" def read_csv_lines(file_name):\n",
- " with open(file_name, 'r') as f:\n",
- " for row in csv.DictReader(f):\n",
+ " with beam_fs.open(file_name) as f:\n",
+ " # Beam reads files as bytes, but csv expects strings,\n",
+ " # so we need to decode the bytes into utf-8 strings.\n",
+ " for row in csv.DictReader(codecs.iterdecode(f, 'utf-8')):\n",
" yield dict(row)\n",
"\n",
" def expand(self, pcollection):\n",
" return (\n",
" pcollection\n",
" | 'Create file patterns' >> beam.Create(self.file_patterns)\n",
- " | 'Expand file patterns' >> beam.FlatMap(glob.glob)\n",
+ " | 'Expand file patterns' >> beam.FlatMap(self.expand_pattern)\n",
Review comment:
Would Beam's `ReadAllFromText` work better? It handles glob matching and utf-8 decoding.
--
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:
users@infra.apache.org
[GitHub] [beam] davidcavazos commented on pull request #14431: [BEAM-10937] Tour of Beam use FileSystems for I/O
Posted by GitBox <gi...@apache.org>.
davidcavazos commented on pull request #14431:
URL: https://github.com/apache/beam/pull/14431#issuecomment-815127832
Hi @udim, friendly ping on reviewing this :)
--
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:
users@infra.apache.org
[GitHub] [beam] udim commented on a change in pull request #14431: [BEAM-10937] Tour of Beam use FileSystems for I/O
Posted by GitBox <gi...@apache.org>.
udim commented on a change in pull request #14431:
URL: https://github.com/apache/beam/pull/14431#discussion_r616963405
##########
File path: examples/notebooks/tour-of-beam/reading-and-writing-data.ipynb
##########
@@ -724,16 +740,21 @@
" self.batch_size = batch_size\n",
"\n",
" @staticmethod\n",
+ " def expand_pattern(pattern):\n",
+ " for match_result in beam_fs.match([pattern])[0].metadata_list:\n",
+ " yield match_result.path\n",
+ "\n",
+ " @staticmethod\n",
" def write_file(lines, file_name_prefix, file_name_suffix):\n",
" file_name = f\"{file_name_prefix}-{uuid.uuid4().hex}{file_name_suffix}\"\n",
- " with open(file_name, 'w') as f:\n",
+ " with beam_fs.create(file_name) as f:\n",
" for line in lines:\n",
- " f.write(f\"{line}\\n\")\n",
+ " f.write(f\"{line}\\n\".encode('utf-8'))\n",
"\n",
" def expand(self, pcollection):\n",
" # Remove existing files matching the output file_name pattern.\n",
- " for f in glob.glob(f\"{self.file_name_prefix}*{self.file_name_suffix}\"):\n",
- " os.remove(f)\n",
+ " for path in self.expand_pattern(f\"{self.file_name_prefix}*{self.file_name_suffix}\"):\n",
Review comment:
Similarly, `WriteToText` might be used to simplify things.
--
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:
users@infra.apache.org
[GitHub] [beam] udim commented on a change in pull request #14431: [BEAM-10937] Tour of Beam use FileSystems for I/O
Posted by GitBox <gi...@apache.org>.
udim commented on a change in pull request #14431:
URL: https://github.com/apache/beam/pull/14431#discussion_r618838488
##########
File path: examples/notebooks/tour-of-beam/reading-and-writing-data.ipynb
##########
@@ -348,36 +396,37 @@
"base_uri": "https://localhost:8080/"
},
"id": "m8iXqE1CRnn5",
- "outputId": "5c9fa281-1a05-4d20-ccdf-7899df6cc083"
+ "outputId": "4f28f0f9-6ba9-4814-ddab-7fed39e64640"
},
"source": [
"import apache_beam as beam\n",
- "\n",
- "class Count(beam.PTransform):\n",
- " def __init__(self, n):\n",
- " self.n = n\n",
- "\n",
- " @staticmethod\n",
- " def count(n):\n",
+ "from apache_beam.options.pipeline_options import PipelineOptions\n",
+ "from typing import Iterable\n",
+ "\n",
+ "@beam.ptransform_fn\n",
+ "@beam.typehints.with_input_types(beam.pvalue.PBegin)\n",
+ "@beam.typehints.with_output_types(int)\n",
+ "def Count(pcollection: beam.PCollection, n: int):\n",
Review comment:
I would put full type annotations here:
```py
"def Count(pcollection: beam.pvalue.PBegin, n: int) -> beam.PCollection[int]:\n",
```
##########
File path: examples/notebooks/tour-of-beam/reading-and-writing-data.ipynb
##########
@@ -348,36 +396,37 @@
"base_uri": "https://localhost:8080/"
},
"id": "m8iXqE1CRnn5",
- "outputId": "5c9fa281-1a05-4d20-ccdf-7899df6cc083"
+ "outputId": "4f28f0f9-6ba9-4814-ddab-7fed39e64640"
},
"source": [
"import apache_beam as beam\n",
- "\n",
- "class Count(beam.PTransform):\n",
- " def __init__(self, n):\n",
- " self.n = n\n",
- "\n",
- " @staticmethod\n",
- " def count(n):\n",
+ "from apache_beam.options.pipeline_options import PipelineOptions\n",
+ "from typing import Iterable\n",
+ "\n",
+ "@beam.ptransform_fn\n",
+ "@beam.typehints.with_input_types(beam.pvalue.PBegin)\n",
+ "@beam.typehints.with_output_types(int)\n",
+ "def Count(pcollection: beam.PCollection, n: int):\n",
Review comment:
I would put full type annotations here:
```python
"def Count(pcollection: beam.pvalue.PBegin, n: int) -> beam.PCollection[int]:\n",
```
--
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:
users@infra.apache.org
[GitHub] [beam] udim commented on a change in pull request #14431: [BEAM-10937] Tour of Beam use FileSystems for I/O
Posted by GitBox <gi...@apache.org>.
udim commented on a change in pull request #14431:
URL: https://github.com/apache/beam/pull/14431#discussion_r618840227
##########
File path: examples/notebooks/tour-of-beam/reading-and-writing-data.ipynb
##########
@@ -606,32 +657,31 @@
" def teardown(self):\n",
" self.connection.close()\n",
"\n",
- "class SelectFromSQLite(beam.PTransform):\n",
- " def __init__(self, database_file, queries):\n",
- " self.database_file = database_file\n",
- " self.queries = queries\n",
- "\n",
- " def expand(self, pcollection):\n",
- " return (\n",
- " pcollection\n",
- " | 'Create None' >> beam.Create(queries)\n",
- " | 'SQLite SELECT' >> beam.ParDo(SQLiteSelect(self.database_file))\n",
- " )\n",
+ "@beam.ptransform_fn\n",
+ "@beam.typehints.with_input_types(beam.pvalue.PBegin)\n",
+ "@beam.typehints.with_output_types(Dict[str, str])\n",
+ "def SelectFromSQLite(pcollection, database_file: str, queries: List[Tuple[str, List[str]]]):\n",
Review comment:
same here and below
--
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:
users@infra.apache.org
[GitHub] [beam] udim merged pull request #14431: [BEAM-10937] Tour of Beam use FileSystems for I/O
Posted by GitBox <gi...@apache.org>.
udim merged pull request #14431:
URL: https://github.com/apache/beam/pull/14431
--
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:
users@infra.apache.org
[GitHub] [beam] udim commented on a change in pull request #14431: [BEAM-10937] Tour of Beam use FileSystems for I/O
Posted by GitBox <gi...@apache.org>.
udim commented on a change in pull request #14431:
URL: https://github.com/apache/beam/pull/14431#discussion_r618838488
##########
File path: examples/notebooks/tour-of-beam/reading-and-writing-data.ipynb
##########
@@ -348,36 +396,37 @@
"base_uri": "https://localhost:8080/"
},
"id": "m8iXqE1CRnn5",
- "outputId": "5c9fa281-1a05-4d20-ccdf-7899df6cc083"
+ "outputId": "4f28f0f9-6ba9-4814-ddab-7fed39e64640"
},
"source": [
"import apache_beam as beam\n",
- "\n",
- "class Count(beam.PTransform):\n",
- " def __init__(self, n):\n",
- " self.n = n\n",
- "\n",
- " @staticmethod\n",
- " def count(n):\n",
+ "from apache_beam.options.pipeline_options import PipelineOptions\n",
+ "from typing import Iterable\n",
+ "\n",
+ "@beam.ptransform_fn\n",
+ "@beam.typehints.with_input_types(beam.pvalue.PBegin)\n",
+ "@beam.typehints.with_output_types(int)\n",
+ "def Count(pcollection: beam.PCollection, n: int):\n",
Review comment:
I would put full type annotations here:
```python
"def Count(pcollection: beam.pvalue.PBegin, n: int) -> beam.PCollection[int]:\n",
```
and possibly rename `pcollection` to `pbegin`.
--
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:
users@infra.apache.org
[GitHub] [beam] davidcavazos commented on a change in pull request #14431: [BEAM-10937] Tour of Beam use FileSystems for I/O
Posted by GitBox <gi...@apache.org>.
davidcavazos commented on a change in pull request #14431:
URL: https://github.com/apache/beam/pull/14431#discussion_r617767804
##########
File path: examples/notebooks/tour-of-beam/reading-and-writing-data.ipynb
##########
@@ -423,28 +422,36 @@
"base_uri": "https://localhost:8080/"
},
"id": "ywVbJxegaZbo",
- "outputId": "5e0adfa3-e685-4fe0-b6b7-bfa3d8469da1"
+ "outputId": "042ee6be-d4d6-4983-9a7c-5f876ba0f655"
},
"source": [
"import apache_beam as beam\n",
+ "from apache_beam.io.filesystems import FileSystems as beam_fs\n",
+ "import codecs\n",
"import csv\n",
- "import glob\n",
"\n",
"class ReadCsvFiles(beam.PTransform):\n",
" def __init__(self, file_patterns):\n",
" self.file_patterns = file_patterns\n",
"\n",
" @staticmethod\n",
+ " def expand_pattern(pattern):\n",
+ " for match_result in beam_fs.match([pattern])[0].metadata_list:\n",
+ " yield match_result.path\n",
+ "\n",
+ " @staticmethod\n",
" def read_csv_lines(file_name):\n",
- " with open(file_name, 'r') as f:\n",
- " for row in csv.DictReader(f):\n",
+ " with beam_fs.open(file_name) as f:\n",
+ " # Beam reads files as bytes, but csv expects strings,\n",
+ " # so we need to decode the bytes into utf-8 strings.\n",
+ " for row in csv.DictReader(codecs.iterdecode(f, 'utf-8')):\n",
" yield dict(row)\n",
"\n",
" def expand(self, pcollection):\n",
" return (\n",
" pcollection\n",
" | 'Create file patterns' >> beam.Create(self.file_patterns)\n",
- " | 'Expand file patterns' >> beam.FlatMap(glob.glob)\n",
+ " | 'Expand file patterns' >> beam.FlatMap(self.expand_pattern)\n",
Review comment:
Yes, it could work, but the intention of this particular sample was on how to create a custom I/O transform. I picked this example since users are already familiar with how it worked so I thought it was simpler. I'll do add a note pointing them to `ReadAllFromText` so they can check it out as well if they need something similar to this.
--
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:
users@infra.apache.org
[GitHub] [beam] davidcavazos commented on a change in pull request #14431: [BEAM-10937] Tour of Beam use FileSystems for I/O
Posted by GitBox <gi...@apache.org>.
davidcavazos commented on a change in pull request #14431:
URL: https://github.com/apache/beam/pull/14431#discussion_r617768208
##########
File path: examples/notebooks/tour-of-beam/reading-and-writing-data.ipynb
##########
@@ -724,16 +740,21 @@
" self.batch_size = batch_size\n",
"\n",
" @staticmethod\n",
+ " def expand_pattern(pattern):\n",
+ " for match_result in beam_fs.match([pattern])[0].metadata_list:\n",
+ " yield match_result.path\n",
+ "\n",
+ " @staticmethod\n",
" def write_file(lines, file_name_prefix, file_name_suffix):\n",
" file_name = f\"{file_name_prefix}-{uuid.uuid4().hex}{file_name_suffix}\"\n",
- " with open(file_name, 'w') as f:\n",
+ " with beam_fs.create(file_name) as f:\n",
" for line in lines:\n",
- " f.write(f\"{line}\\n\")\n",
+ " f.write(f\"{line}\\n\".encode('utf-8'))\n",
"\n",
" def expand(self, pcollection):\n",
" # Remove existing files matching the output file_name pattern.\n",
- " for f in glob.glob(f\"{self.file_name_prefix}*{self.file_name_suffix}\"):\n",
- " os.remove(f)\n",
+ " for path in self.expand_pattern(f\"{self.file_name_prefix}*{self.file_name_suffix}\"):\n",
Review comment:
Again, this was more to show how to create custom I/O transforms, but I'll definitely add a note pointing them to `WriteToText`.
--
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:
users@infra.apache.org