Repository: dask/dask
Pull request #7394 (2021-03-15 14:32:03)

DOC hyperlink repartition
Not tested


Patch:
diff --git a/dask/bag/text.py b/dask/bag/text.py
index 11200a57d..b9a9a3e63 100644
--- a/dask/bag/text.py
+++ b/dask/bag/text.py
@@ -1,5 +1,4 @@
 import io
-import os
 from functools import partial
 
 from tlz import concat
@@ -19,7 +18,7 @@ def read_text(
     compression="infer",
     encoding=system_encoding,
     errors="strict",
-    linedelimiter=os.linesep,
+    linedelimiter=None,
     collection=True,
     storage_options=None,
     files_per_partition=None,
@@ -42,7 +41,7 @@ def read_text(
         Compression format like 'gzip' or 'xz'.  Defaults to 'infer'
     encoding: string
     errors: string
-    linedelimiter: string
+    linedelimiter: string or None
     collection: bool, optional
         Return dask.bag if True, or list of delayed values if false
     storage_options: dict
@@ -90,18 +89,26 @@ def read_text(
     if isinstance(blocksize, str):
         blocksize = parse_bytes(blocksize)
 
-    files = open_files(
-        urlpath,
-        mode="rt",
-        encoding=encoding,
-        errors=errors,
-        compression=compression,
-        **(storage_options or {})
-    )
     if blocksize is None:
+        if linedelimiter in [None, "", "\n", "\r", "\r\n"]:
+            newline = linedelimiter
+            linedelimiter = None
+        else:
+            newline = ""
+        files = open_files(
+            urlpath,
+            mode="rt",
+            encoding=encoding,
+            errors=errors,
+            compression=compression,
+            newline=newline,
+            **(storage_options or {})
+        )
         if files_per_partition is None:
             blocks = [
-                delayed(list)(delayed(partial(file_to_blocks, include_path))(fil))
+                delayed(partial(file_to_blocks, include_path, delimiter=linedelimiter))(
+                    fil
+                )
                 for fil in files
             ]
         else:
@@ -110,15 +117,18 @@ def read_text(
                 block_files = files[start : (start + files_per_partition)]
                 block_lines = delayed(concat)(
                     delayed(map)(
-                        partial(file_to_blocks, include_path),
+                        partial(file_to_blocks, include_path, delimiter=linedelimiter),
                         block_files,
                     )
                 )
                 blocks.append(block_lines)
     else:
+        # special case for linedelimiter=None: we will need to split on an actual bytestring
+        # and the line reader will then use "universal" mode. Just as well that \r\n and \n
+        # will both work (thankfully \r for MacOS is no longer a thing)
         o = read_bytes(
             urlpath,
-            delimiter=linedelimiter.encode(),
+            delimiter=linedelimiter.encode() if linedelimiter is not None else b"\n",
             blocksize=blocksize,
             sample=False,
             compression=compression,
@@ -126,7 +136,10 @@ def read_text(
             **(storage_options or {})
         )
         raw_blocks = o[1]
-        blocks = [delayed(decode)(b, encoding, errors) for b in concat(raw_blocks)]
+        blocks = [
+            delayed(decode)(b, encoding, errors, linedelimiter)
+            for b in concat(raw_blocks)
+        ]
         if include_path:
             paths = list(
                 concat([[path] * len(raw_blocks[i]) for i, path in enumerate(o[2])])
@@ -144,10 +157,21 @@ def read_text(
     return blocks
 
 
-def file_to_blocks(include_path, lazy_file):
+def file_to_blocks(include_path, lazy_file, delimiter=None):
+    # blocksize is None branch
     with lazy_file as f:
-        for line in f:
-            yield (line, lazy_file.path) if include_path else line
+        if delimiter is not None:
+            text = f.read()
+            if not text:
+                return []
+            parts = text.split(delimiter)
+            yield from (
+                (line, lazy_file.path) if include_path else line
+                for line in [line + delimiter for line in parts[:-1]] + parts[-1:]
+            )
+        else:
+            for line in f:
+                yield (line, lazy_file.path) if include_path else line
 
 
 def attach_path(block, path):
@@ -155,7 +179,17 @@ def attach_path(block, path):
         yield (p, path)
 
 
-def decode(block, encoding, errors):
+def decode(block, encoding, errors, line_delimiter):
+    # blocksize is not None branch
     text = block.decode(encoding, errors)
-    lines = io.StringIO(text)
-    return list(lines)
+    if line_delimiter in [None, "", "\n", "\r", "\r\n"]:
+        lines = io.StringIO(text, newline=line_delimiter)
+        return list(lines)
+    else:
+        if not text:
+            return []
+        parts = text.split(line_delimiter)
+        out = [t + line_delimiter for t in parts[:-1]] + (
+            parts[-1:] if not text.endswith(line_delimiter) else []
+        )
+        return out
diff --git a/docs/source/dataframe-best-practices.rst b/docs/source/dataframe-best-practices.rst
index ab4df7fe7..42d3528d1 100644
--- a/docs/source/dataframe-best-practices.rst
+++ b/docs/source/dataframe-best-practices.rst
@@ -162,7 +162,7 @@ A common situation is that you load lots of data into reasonably sized
 partitions (Dask's defaults make decent choices), but then you filter down your
 dataset to only a small fraction of the original.  At this point, it is wise to
 regroup your many small partitions into a few larger ones.  You can do this by
-using the ``repartition`` method:
+using the :py:class:`dask.dataframe.DataFrame.repartition` method:
 
 .. code-block:: python
 


Test patch:
diff --git a/dask/bag/tests/test_text.py b/dask/bag/tests/test_text.py
index 2215bc0e2..d89d68f6b 100644
--- a/dask/bag/tests/test_text.py
+++ b/dask/bag/tests/test_text.py
@@ -119,3 +119,21 @@ def test_errors():
         result = read_text(".test.foo", encoding="ascii", errors="ignore")
         result = result.compute(scheduler="sync")
         assert result == ["Jos\n", "Alice"]
+
+
+def test_complex_delimiter():
+    longstr = "abc\ndef\n123\n$$$$\ndog\ncat\nfish\n\n\r\n$$$$hello"
+    with filetexts({".test.delim.txt": longstr}):
+        assert read_text(".test.delim.txt", linedelimiter="$$$$").count().compute() == 3
+        assert (
+            read_text(".test.delim.txt", linedelimiter="$$$$", blocksize=2)
+            .count()
+            .compute()
+            == 3
+        )
+        vals = read_text(".test.delim.txt", linedelimiter="$$$$").compute()
+        assert vals[-1] == "hello"
+        assert vals[0].endswith("$$$$")
+        vals = read_text(".test.delim.txt", linedelimiter="$$$$", blocksize=2).compute()
+        assert vals[-1] == "hello"
+        assert vals[0].endswith("$$$$")