v2.1.15 dataflows>=0.0.65 with deduplicate processor

Author: akariv

README.md

@@ -740,6 +740,27 @@ Filtering just American and European countries, leaving out countries whose main
     sort-by: "{country_name}"
 ```
 
+### ***`deduplicate`***
+
+Deduplicates rows in resources based on the resources' primary key
+
+`deduplicate` accepts a resource specifier - for each resource, it will output only unique rows (based on the values in the primary key fields). Rows with duplicate primary keys will be ignored.
+
+_Parameters_:
+
+- `resources` - Which resources to sort. Same semantics as `resources` in `stream_remote_resources`.
+
+*Examples*:
+
+Deduplicating rows in the `world-population` resource.
+
+```yaml
+- run: deduplicate
+  parameters:
+    resources: world_population
+```
+
+
 ### ***`duplicate`***
 
 Duplicate a resource.

datapackage_pipelines/VERSION

@@ -1 +1 @@
-2.1.14
+2.1.15

datapackage_pipelines/lib/deduplicate.py

@@ -0,0 +1,16 @@
+from dataflows import Flow, deduplicate
+from datapackage_pipelines.wrapper import ingest
+from datapackage_pipelines.utilities.flow_utils import spew_flow
+
+
+def flow(parameters):
+    return Flow(
+        deduplicate(
+            resources=parameters.get('resources'),
+        )
+    )
+
+
+if __name__ == '__main__':
+    with ingest() as ctx:
+        spew_flow(flow(ctx.parameters), ctx)

setup.py

@@ -39,7 +39,7 @@ def read(*paths):
     'cachetools',
     'tabulator>=1.17.0',
     'globster>=0.1.0',
-    'dataflows>=0.0.57',
+    'dataflows>=0.0.65',
     'python-dateutil<2.8.1',
 ]
 SPEEDUP_REQUIRES = [

tests/stdlib/fixtures/simple_deduplicate

@@ -0,0 +1,109 @@
+deduplicate
+--
+{
+    "resources": ["concat-a1", "concat-a2"]
+}
+--
+{
+    "name": "test",
+    "resources": [
+        {
+            "name": "concat-a1",
+            "dpp:streaming": true,
+            "path": "concat-a1.csv",
+            "schema": { "fields": [
+                {"name": "a1", "type": "string"},
+                {"name": "a2", "type": "string"},
+                {"name": "a3", "type": "string"}
+            ], "primaryKey": ["a1", "a2"]}
+        },
+        {
+            "name": "concat-a2",
+            "dpp:streaming": true,
+            "path": "concat-a2.csv",
+            "schema": { "fields": [
+                {"name": "a1", "type": "string"},
+                {"name": "a2", "type": "string"},
+                {"name": "a3", "type": "string"}
+            ]}
+        },
+        {
+            "name": "concat-c",
+            "dpp:streaming": true,
+            "path": "concat-c.csv",
+            "schema": { "fields": [
+                {"name": "c1", "type": "string"},
+                {"name": "c2", "type": "string"},
+                {"name": "c3", "type": "string"}
+            ]}
+        }
+    ]
+}
+--
+{"a1":"a1","a2":"a1","a3":"a2"}
+{"a1":"a2","a2":"a1","a3":"a1"}
+{"a1":"a1","a2":"a1","a3":"a2"}
+{"a1":"a2","a2":"a1","a3":"a1"}
+
+{"a1":"a1","a2":"a3","a3":"a2"}
+{"a1":"a2","a2":"a3","a3":"a1"}
+{"a1":"a3","a2":"a4","a3":"a2"}
+{"a1":"a4","a2":"a4","a3":"a1"}
+
+{"c1":"c11","c2":"c21","c3":"c31"}
+{"c1":"c12","c2":"c22","c3":"c32"}
+{"c1":"c13","c2":"c23","c3":"c33"}
+--
+{
+    "name": "test",
+    "profile": "data-package",
+    "resources": [
+        {
+            "name": "concat-a1",
+            "dpp:streaming": true,
+            "path": "concat-a1.csv",
+            "profile": "data-resource",
+            "schema": { "fields": [
+                {"name": "a1", "type": "string"},
+                {"name": "a2", "type": "string"},
+                {"name": "a3", "type": "string"}
+            ], "primaryKey": ["a1", "a2"]}
+        },
+        {
+            "name": "concat-a2",
+            "dpp:streaming": true,
+            "path": "concat-a2.csv",
+            "profile": "data-resource",
+            "schema": { "fields": [
+                {"name": "a1", "type": "string"},
+                {"name": "a2", "type": "string"},
+                {"name": "a3", "type": "string"}
+            ]}
+        },
+        {
+            "name": "concat-c",
+            "dpp:streaming": true,
+            "path": "concat-c.csv",
+            "profile": "data-resource",
+            "schema": { "fields": [
+                {"name": "c1", "type": "string"},
+                {"name": "c2", "type": "string"},
+                {"name": "c3", "type": "string"}
+            ]}
+        }
+    ]
+}
+--
+{"a1":"a1","a2":"a1","a3":"a2"}
+{"a1":"a2","a2":"a1","a3":"a1"}
+
+{"a1":"a1","a2":"a3","a3":"a2"}
+{"a1":"a2","a2":"a3","a3":"a1"}
+{"a1":"a3","a2":"a4","a3":"a2"}
+{"a1":"a4","a2":"a4","a3":"a1"}
+
+{"c1":"c11","c2":"c21","c3":"c31"}
+{"c1":"c12","c2":"c22","c3":"c32"}
+{"c1":"c13","c2":"c23","c3":"c33"}
+
+{}