Minor

.gitignore

@@ -5,6 +5,7 @@ build
 dist
 env
 initdb.py
+dbinit.py
 logs
 *.cfg
 MANIFEST

airflow/bin/airflow

@@ -228,6 +228,7 @@ def initdb(args):
 
 if __name__ == '__main__':
 
+    reload(logging)
     logging.root.handlers = []
 
     parser = argparse.ArgumentParser()

airflow/executors/local_executor.py

@@ -45,7 +45,7 @@ def run(self):
 
 class LocalExecutor(BaseExecutor):
 
-    def __init__(self, parallelism=4):
+    def __init__(self, parallelism=16):
         super(LocalExecutor, self).__init__()
         self.parallelism = parallelism
 

airflow/models.py

@@ -377,8 +377,7 @@ def run(
             self, verbose=True,
             ignore_dependencies=False,
             force=False,
-            mark_success=False,
-            ):
+            mark_success=False,):
         """
         Runs the task instnace.
         """
@@ -601,7 +600,6 @@ def run(self, dag, start_date=None, end_date=None, mark_success=False):
                 if ti.state == State.SUCCESS and key in task_instances:
                     del task_instances[key]
                 elif ti.is_runnable():
-                    print "Runnable: " + ti.task_id
                     executor.queue_command(
                         key=ti.key, command=ti.command(
                             mark_success=mark_success,
@@ -1101,7 +1099,12 @@ def add_tasks(self, tasks):
             self.add_task(task)
 
     def db_merge(self):
+        BO = BaseOperator
         session = settings.Session()
+        tasks = session.query(BO).filter(BO.dag_id==self.dag_id).all()
+        for t in tasks:
+            session.delete(t)
+        session.commit()
         session.merge(self)
         session.commit()
 

airflow/settings.py

@@ -22,7 +22,7 @@
 Session = sessionmaker()
 #engine = create_engine('mysql://airflow:airflow@localhost/airflow')
 engine = create_engine(
-    SQL_ALCHEMY_CONN, pool_size=25)
+    SQL_ALCHEMY_CONN, pool_size=50)
 Session.configure(bind=engine)
 
 # can't move this to configuration due to ConfigParser interpolation