This is an automated email from the ASF dual-hosted git repository.
jrmccluskey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 539b048eb8d Fix dataframe CSV tests on Windows (#39563)
539b048eb8d is described below
commit 539b048eb8d0fdce1af2faaed310728b07d89832
Author: Guflly <[email protected]>
AuthorDate: Mon Aug 3 07:22:34 2026 -0700
Fix dataframe CSV tests on Windows (#39563)
---
sdks/python/apache_beam/dataframe/io.py | 2 +-
sdks/python/apache_beam/dataframe/io_test.py | 15 ++++++++-------
2 files changed, 9 insertions(+), 8 deletions(-)
diff --git a/sdks/python/apache_beam/dataframe/io.py
b/sdks/python/apache_beam/dataframe/io.py
index 21eab0b82fa..bc39a40403f 100644
--- a/sdks/python/apache_beam/dataframe/io.py
+++ b/sdks/python/apache_beam/dataframe/io.py
@@ -736,7 +736,7 @@ class _WriteToPandasFileSink(fileio.FileSink):
self.empty = self.header = self.footer = None
if not self.binary:
file_handle = TextIOWrapper(
- file_handle, encoding=self.kwargs.get("encoding", None))
+ file_handle, encoding=self.kwargs.get("encoding", None), newline='')
self.file_handle = file_handle
def write_to(self, df, file_handle=None):
diff --git a/sdks/python/apache_beam/dataframe/io_test.py
b/sdks/python/apache_beam/dataframe/io_test.py
index 4cd502d1b8d..dd7b8db497c 100644
--- a/sdks/python/apache_beam/dataframe/io_test.py
+++ b/sdks/python/apache_beam/dataframe/io_test.py
@@ -18,7 +18,6 @@ import glob
import importlib
import math
import os
-import platform
import shutil
import tempfile
import typing
@@ -65,9 +64,6 @@ class MyRow(typing.NamedTuple):
value: int
[email protected](
- platform.system() == 'Windows',
- 'https://github.com/apache/beam/issues/20642')
class IOTest(unittest.TestCase):
def setUp(self):
self._temp_roots = []
@@ -431,6 +427,11 @@ X , c1, c2
def test_windowed_write(self):
output = self.temp_dir()
+
+ def no_colon_file_naming(*args):
+ file_name = fileio.default_file_naming('out.csv')(*args)
+ return file_name.replace(':', '_')
+
with beam.Pipeline() as p:
pc = (
p | beam.Create([MyRow(timestamp=i, value=i % 3) for i in range(20)])
@@ -440,18 +441,18 @@ X , c1, c2
beam.window.FixedWindows(10)).with_output_types(MyRow))
deferred_df = convert.to_dataframe(pc)
- deferred_df.to_csv(output + 'out.csv', index=False)
+ deferred_df.to_csv(output, file_naming=no_colon_file_naming, index=False)
first_window_files = (
f'{output}out.csv-'
- f'{datetime.utcfromtimestamp(0).isoformat()}*')
+ f'{datetime.utcfromtimestamp(0).isoformat().replace(":", "_")}*')
self.assertCountEqual(
['timestamp,value'] + [f'{i},{i % 3}' for i in range(10)],
set(self.read_all_lines(first_window_files, delete=True)))
second_window_files = (
f'{output}out.csv-'
- f'{datetime.utcfromtimestamp(10).isoformat()}*')
+ f'{datetime.utcfromtimestamp(10).isoformat().replace(":", "_")}*')
self.assertCountEqual(
['timestamp,value'] + [f'{i},{i%3}' for i in range(10, 20)],
set(self.read_all_lines(second_window_files, delete=True)))
