Index documents as well

Author: akariv

datapackage_pipelines_budgetkey/generator.py

@@ -64,6 +64,21 @@ def generate_pipeline(cls, source):
                                  'doc_type': doc_type}
                             ]
                         }
+                    }),
+                    ('convert_to_key_value', {
+                        'key-prefix': doc_type,
+                        'key-fields': key_fields
+                    }),
+                    ('sample'
+
+                    ),
+                    ('dump_to_es', {
+                        'indexes': {
+                            'budgetkey': [
+                                {'resource-name': 'document',
+                                 'doc_type': 'document'}
+                            ]
+                        }
                     })
                 ]
 

datapackage_pipelines_budgetkey/pipelines/budgetkey/pipeline-spec.yaml

@@ -26,6 +26,7 @@ budget-functional-aggregates:
         target:
           name: budget-functional-aggregates
           key: null
+
         fields:
           func_cls_title_1: null
           func_cls_title_2: null

datapackage_pipelines_budgetkey/processors/convert_to_key_value.py

@@ -0,0 +1,32 @@
+from datapackage_pipelines.utilities.resources import PROP_STREAMING
+from datapackage_pipelines.wrapper import ingest, spew
+
+parameters, dp, res_iter = ingest()
+
+prefix = parameters['key-prefix']
+key_fields = parameters['key-fields']
+
+def process_resources(res_iter_):
+    for res in res_iter_:
+        for row in res:
+            yield {
+                'key': '/'.join([prefix] + [str(row[k]) for k in key_fields]),
+                'value': dict(row)
+            }
+
+dp['resources'] = [
+    {
+        'name': 'document',
+        PROP_STREAMING: True,
+        'path': 'data/documents.csv',
+        'schema': {
+            'fields': [
+                {'name': 'key', 'type': 'string'},
+                {'name': 'value', 'type': 'object', 'es:index': False}
+            ],
+            'primaryKey': ['key']
+        }
+    }
+]
+
+spew(dp, [process_resources(res_iter)])