start date

snippets/pyaloha/main.py

@@ -6,7 +6,7 @@
 import os
 import argparse
 
-from pyaloha.protocol.base import str2date
+from pyaloha.protocol import WorkerResults, str2date
 from pyaloha.settings import DEFAULT_WORKER_NUM, DEFAULT_ALOHA_DATA_DIR
 from pyaloha.worker import invoke_cmd_worker, load_plugin, setup_logs
 
@@ -103,7 +103,7 @@ def aggregate_raw_data(
     ]
 
     tasks = [
-        (plugin_dir, plugin, fpath, events_limit)
+        (plugin_dir, plugin, fpath, events_limit, start_date)
         for fpath in files
     ]
 
@@ -135,7 +135,7 @@ def aggregate_raw_data(
             )
             for file_name, results in engine(invoke_cmd_worker, batch_tasks):
                 try:
-                    results = aggregator.loads_results(results)
+                    results = WorkerResults.loads_object(results)
                     logger.info(
                         'Aggregator: task %s is being aggregated' % file_name
                     )

snippets/pyaloha/worker.py

@@ -29,10 +29,10 @@ def invoke_cmd_worker(item):
     try:
         pid = multiprocessing.current_process().pid
 
-        plugin_dir, plugin, filepath, events_limit = item
+        plugin_dir, plugin, filepath, events_limit, start_date = item
         worker_fpath = os.path.abspath(__file__)
-        cmd = 'gzip -d -c %s | python2.7 %s %s %s %s' % (
-            filepath, worker_fpath, plugin_dir, plugin, events_limit
+        cmd = 'gzip -d -c %s | python2.7 %s %s %s %s %s' % (
+            filepath, worker_fpath, plugin_dir, plugin, events_limit, start_date
         )
         logger.info(
             '%d: Starting job: %s', pid, cmd
@@ -71,6 +71,10 @@ def parse_args():
         help='Read only first N events. If not specified, will read all events',
         default=0
     )
+    parser.add_argument(
+        'start_date',
+        help='First date of tracking range',
+    )
     return parser.parse_args()
 
 
@@ -82,7 +86,7 @@ def worker():
     try:
         processor = load_plugin(
             args.plugin, plugin_dir=args.plugin_dir
-        ).DataStreamWorker()
+        ).DataStreamWorker(start_date=args.start_date)
         iterate_events(processor, events_limit=args.events_limit)
         processor.pre_output()
         sys.stdout.write(processor.dumps_results() + '\n')