diff --git a/pipelib/core.py b/pipelib/core.py
index 36f7b37..ec31d3e 100644
--- a/pipelib/core.py
+++ b/pipelib/core.py
@@ -80,17 +80,17 @@ class Dataset:
             yield from chain(dataset, *others)
         return PipelinedDataset(self, f)
 
-    def map_parallel(self, map_func, n=None, chunksize=1):
+    def map_parallel(self, map_func, n=None, chunksize=1, unordered=False):
         return PipelinedDataset(
-            self, parallel.MapParallel(map_func, n, chunksize))
+            self, parallel.MapParallel(map_func, n, chunksize, unordered))
 
-    def flat_map_parallel(self, map_func, n=None, chunksize=1):
+    def flat_map_parallel(self, map_func, n=None, chunksize=1, unordered=False):
         return PipelinedDataset(
-            self, parallel.FlatMapParallel(map_func, n, chunksize))
+            self, parallel.FlatMapParallel(map_func, n, chunksize, unordered))
 
-    def filter_parallel(self, predicate, n=None, chunksize=1):
+    def filter_parallel(self, predicate, n=None, chunksize=1, unordered=False):
         return PipelinedDataset(
-            self, parallel.FilterParallel(predicate, n, chunksize))
+            self, parallel.FilterParallel(predicate, n, chunksize, unordered))
 
     def all(self):
         return list(self)
diff --git a/pipelib/parallel.py b/pipelib/parallel.py
index 98dfdd8..c7433ee 100644
--- a/pipelib/parallel.py
+++ b/pipelib/parallel.py
@@ -4,21 +4,27 @@ import multiprocess
 
 
 class MapParallel:
-    def __init__(self, func, n=None, chunksize=1):
+    def __init__(self, func, n=None, chunksize=1, unordered=False):
         self._func = func
         self._n = n
         self._chunksize = chunksize
+        if not unordered:
+            self._map_method = 'imap'
+        else:
+            self._map_method = 'imap_unordered'
 
     def __call__(self, dataset):
         with multiprocess.Pool(self._n) as p:
-            yield from p.imap_unordered(self._func, dataset, self._chunksize)
+            yield from getattr(p, self._map_method)(
+                self._func, dataset, self._chunksize)
 
 
 class FlatMapParallel(MapParallel):
     def __call__(self, dataset):
         with multiprocess.Pool(self._n) as p:
             yield from chain.from_iterable(
-                p.imap_unordered(self._func, dataset, self._chunksize))
+                getattr(p, self._map_method)(
+                    self._func, dataset, self._chunksize))
 
 
 class FilterParallel(MapParallel):
@@ -37,4 +43,5 @@ class FilterParallel(MapParallel):
 
         with multiprocess.Pool(self._n) as p:
             yield from (x for x, keep in
-                        p.imap_unordered(task, dataset, self._chunksize) if keep)
+                        getattr(p, self._map_method)(
+                            task, dataset, self._chunksize) if keep)
diff --git a/setup.py b/setup.py
index eba6daf..0854f8a 100644
--- a/setup.py
+++ b/setup.py
@@ -7,7 +7,7 @@ except ImportError:
 
 setup(
     name='pipelib',
-    version='0.2.2',
+    version='0.2.3',
     description='pipeline architecture data library',
     long_description=open('./README.md', encoding='utf-8').read(),
     long_description_content_type='text/markdown',

Option for order/unorder
- [x] `Dataset.map_parallel`
- [x] `Dataset.flat_map_parallel`
- [x] `Dataset.filter_parallel`
diff --git a/tests/test_parallel.py b/tests/test_parallel.py
index c65ddc8..5502153 100644
--- a/tests/test_parallel.py
+++ b/tests/test_parallel.py
@@ -9,24 +9,45 @@ class ParallelTestCase(TestCase):
         self.data = range(100)
 
     def test_map_parallel(self):
-        result = list(parallel.MapParallel(lambda x: x ** 2)(self.data))
-        result.sort()
         expected = [x ** 2 for x in self.data]
+        # ordered
+        result = parallel.MapParallel(lambda x: x ** 2)(self.data)
+        for x, y in zip(result, expected):
+            self.assertEqual(x, y)
+        # unordered
+        result = list(parallel.MapParallel(
+            lambda x: x ** 2, unordered=True)(self.data))
+        result.sort()
         self.assertListEqual(result, expected)
 
     def test_filter_parallel(self):
         def predicate(x):
             return x % 2 == 0
 
-        result = list(parallel.FilterParallel(predicate)(self.data))
-        result.sort()
         task = parallel.FilterParallel._FilterTask(predicate)
-        expected = [task(x) for x in self.data]
-        expected = [x[0] for x in expected if x[1]]
+        expected = [task(x)[0] for x in self.data if task(x)[1]]
+
+        # ordered
+        result = parallel.FilterParallel(predicate)(self.data)
+        for x, y in zip(result, expected):
+            self.assertEqual(x, y)
+
+        # unordered
+        result = list(parallel.FilterParallel(
+            predicate, unordered=True)(self.data))
+        result.sort()
         self.assertListEqual(result, expected)
 
     def test_flat_map_parallel(self):
-        result = list(parallel.FlatMapParallel(lambda x: [x])(self.data))
-        result.sort()
         expected = [x for x in self.data]
+
+        # ordered
+        result = parallel.FlatMapParallel(lambda x: [x])(self.data)
+        for x, y in zip(result, expected):
+            self.assertEqual(x, y)
+
+        # unordered
+        result = list(parallel.FlatMapParallel(
+            lambda x: [x], unordered=True)(self.data))
+        result.sort()
         self.assertListEqual(result, expected)

pip install -e .[dev]
parse_log_pytest
Pipfile
pytest
3.7
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
certifi @ file:///croot/certifi_1671487769961/work/certifi
cloudpickle==2.2.1
dill==0.3.7
exceptiongroup==1.2.2
importlib-metadata==6.7.0
iniconfig==2.0.0
multiprocess==0.70.15
packaging==24.0
-e git+https://github.com/yasufumy/pipelib.git@817dc34a6be1568b6e40b51922b71f36aa5b2c31#egg=pipelib
pipfile==0.0.2
pluggy==1.2.0
pytest==7.4.4
toml @ file:///tmp/build/80754af9/toml_1616166611790/work
tomli==2.0.1
typing_extensions==4.7.1
zipp==3.15.0

name: pipelib
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2022.12.7=py37h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.4.4=h6a678d5_1
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=1.1.1w=h7f8727e_0
  - pip=22.3.1=py37h06a4308_0
  - pipfile=0.0.2=py_0
  - python=3.7.16=h7a1cb2a_0
  - readline=8.2=h5eee18b_0
  - setuptools=65.6.3=py37h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - toml=0.10.2=pyhd3eb1b0_0
  - wheel=0.38.4=py37h06a4308_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - cloudpickle==2.2.1
      - dill==0.3.7
      - exceptiongroup==1.2.2
      - importlib-metadata==6.7.0
      - iniconfig==2.0.0
      - multiprocess==0.70.15
      - packaging==24.0
      - pluggy==1.2.0
      - pytest==7.4.4
      - tomli==2.0.1
      - typing-extensions==4.7.1
      - zipp==3.15.0
prefix: /opt/conda/envs/pipelib

tests/test_parallel.py::ParallelTestCase::test_filter_parallel tests/test_parallel.py::ParallelTestCase::test_flat_map_parallel tests/test_parallel.py::ParallelTestCase::test_map_parallel
setup.py pipelib/core.py pipelib/parallel.py
setup.py pipelib/core.py pipelib/parallel.py