https://linen.dev logo
Join Slack
Powered by
```meltano invoke airflow scheduler -D ...
# announcements
n
Copy code
meltano invoke airflow scheduler -D                                                                                                                                                  
[2020-07-05 15:00:36,336|I|MainThread|alembic.runtime.migration] Context impl SQLiteImpl.
[2020-07-05 15:00:36,337|I|MainThread|alembic.runtime.migration] Will assume non-transactional DDL.
System database up-to-date.
[2020-07-05 15:00:36,955|E|MainThread|root] Meltano could not parse the `discovery.yml` returned by the server.
[2020-07-05 15:00:37,669|E|MainThread|root] Meltano could not parse the `discovery.yml` returned by the server.
[2020-07-05 15:00:43,361|W|MainThread|root] before_configure hook 'before_configure' has failed: airflow initdb failed
[2020-07-05 15:00:43,362|E|MainThread|root] airflow initdb failed
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/meltano/cli/invoke.py", line 32, in invoke
    service = invoker_factory(project, plugin, prepare_with_session=session)
  File "/usr/local/lib/python3.7/site-packages/meltano/core/plugin_invoker.py", line 27, in invoker_factory
    invoker.prepare(prepare_with_session)
  File "/usr/local/lib/python3.7/site-packages/meltano/core/plugin_invoker.py", line 96, in prepare
    with self.plugin.trigger_hooks("configure", self, session):
  File "/usr/local/Cellar/python/3.7.8/Frameworks/Python.framework/Versions/3.7/lib/python3.7/contextlib.py", line 112, in __enter__
    return next(self.gen)
  File "/usr/local/lib/python3.7/site-packages/meltano/core/behavior/hookable.py", line 68, in trigger_hooks
    self.__class__.trigger(self, f"before_{hook_name}", *args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/meltano/core/behavior/hookable.py", line 93, in trigger
    raise err
  File "/usr/local/lib/python3.7/site-packages/meltano/core/behavior/hookable.py", line 87, in trigger
    hook_func(target, *args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/meltano/core/plugin/airflow.py", line 122, in before_configure
    raise SubprocessError("airflow initdb failed", handle)
meltano.core.error.SubprocessError: airflow initdb failed
An error occured: airflow initdb failed.
Aborted!
Is there a way I can get detailed logs? Dont know how to fix airflow initdb cmd thorugh meltano
d
@nehil_jain Can you run again with 
meltano --log-level=debug ...
? That flag isn’t currently documented (which I’ll fix), but that may show more info
In this case I can definitely make the error show automatically though. And let me know if I can help figure out a resolution once you’ve uncovered the error message!
Oh and are you on the latest version of Meltano, by the way?
@nehil_jain Did you manage to resolve this issue?
n
nope, the logs werent useful. I am thinking of moving it to a proper db instead of sqlite. today
d
That's probably a good idea anyway 🙂 I'd love to see the logs though, even if they weren't useful to you, to see if this is actually a bug in Meltano or if something else is going on
@nehil_jain Are you actually on the latest version of Meltano by the way? The "Meltano could not parse the 
discovery.yml
returned by the server." in the output suggests otherwise 😉
n
latest meltano fixed it
d
Awesome!
n
sad
Copy code
[2020-07-06 16:15:37,974] {models.py:273} INFO - Filling up the DagBag from /Users/nehiljain/snaptravel/universal-elt-pipeline/airflow-elt/orchestrate/dags
[2020-07-06 16:15:39 -0400] [70803] [INFO] Handling signal: ttou
[2020-07-06 16:15:39 -0400] [70810] [INFO] Worker exiting (pid: 70810)
d
@nehil_jain You actually have a 
meltano.py
in 
orchestrate/dags
?
n
i need to create one?
d
One should have been added automatically when you ran 
meltano add orchestrator airflow
, but if you were on an old version of Meltano that may not have happened. Try running 
meltano add files airflow
to install the file bundle 🙂
n
I dont
aah
I reran 
meltano add orchestrator airflow
and now i have it
d
Great!
n
Copy code
[2020-07-06 16:21:01,607] {meltano.py:51} INFO - No DAG created for schedule 'gitlab-to-postgres' because it hasn't had a successful (manual) run yet.
[2020-07-06 16:21:01,694] {meltano.py:41} INFO - Considering schedule 'gitlab-to-postgres': {'name': 'gitlab-to-postgres', 'extractor': 'tap-github', 'loader': 'target-postgres', 'transform': 'skip', 'interval': '@hourly', 'start_date': '2020-07-05', 'env': {}, 'cron_interval': '0 * * * *', 'last_successful_run_ended_at': None, 'elt_args': ['tap-github', 'target-postgres', '--transform=skip', '--job_id=gitlab-to-postgres']}
the manual run takes a long time and i want airflow to do the first run for me
d
Feel free to edit 
orchestrate/dags/meltano.py
however you please, including to remove that "manual run" check 🙂
Issue created to remove this behavior from the template: https://gitlab.com/meltano/files-airflow/-/issues/1
n
https://share.getcloudapp.com/rRulzGng there is progress but next hurdle is that my task is not running file the dag is in running state
d
Hmm, now we're getting into "debugging Airflow" territory and I'm less confident I can help 🙂
Meltano isn't doing anything special with Airflow besides configuring it to look for DAGs in that 
orchestrate/dags
directory, fwiw
n
i see
Copy code
[2020-07-06 16:47:29,149] {cli.py:520} INFO - Running <TaskInstance: meltano_gitlab-to-postgres.extract_load 2020-07-06T20:27:58.959483+00:00 [queued]> on host ip-192-168-2-75.ec2.internal
Traceback (most recent call last):
  File "/Users/nehiljain/snaptravel/universal-elt-pipeline/airflow-elt/.meltano/orchestrators/airflow/venv/bin/airflow", line 32, in <module>
    args.func(args)
  File "/Users/nehiljain/snaptravel/universal-elt-pipeline/airflow-elt/.meltano/orchestrators/airflow/venv/lib/python3.7/site-packages/airflow/utils/cli.py", line 74, in wrapper
    return f(*args, **kwargs)
  File "/Users/nehiljain/snaptravel/universal-elt-pipeline/airflow-elt/.meltano/orchestrators/airflow/venv/lib/python3.7/site-packages/airflow/bin/cli.py", line 526, in run
    _run(args, dag, ti)
  File "/Users/nehiljain/snaptravel/universal-elt-pipeline/airflow-elt/.meltano/orchestrators/airflow/venv/lib/python3.7/site-packages/airflow/bin/cli.py", line 440, in _run
    run_job.run()
  File "/Users/nehiljain/snaptravel/universal-elt-pipeline/airflow-elt/.meltano/orchestrators/airflow/venv/lib/python3.7/site-packages/airflow/jobs.py", line 202, in run
    self._execute()
  File "/Users/nehiljain/snaptravel/universal-elt-pipeline/airflow-elt/.meltano/orchestrators/airflow/venv/lib/python3.7/site-packages/airflow/jobs.py", line 2497, in _execute
    self.task_runner = get_task_runner(self)
  File "/Users/nehiljain/snaptravel/universal-elt-pipeline/airflow-elt/.meltano/orchestrators/airflow/venv/lib/python3.7/site-packages/airflow/task/task_runner/__init__.py", line 43, in get_task_runner
    raise AirflowException("Unknown task runner type {}".format(_TASK_RUNNER))
airflow.exceptions.AirflowException: Unknown task runner type StandardTaskRunner
[2020-07-06 16:47:29,947] {sequential_executor.py:52} ERROR - Failed to execute task Command 'airflow run meltano_gitlab-to-postgres extract_load 2020-07-06T20:27:58.959483+00:00 --local -sd /Users/nehiljain/snaptravel/universal-elt-pipeline/airflow-elt/orchestrate/dags/meltano.py' returned non-zero exit status 1..
[2020-07-06 16:47:29,948] {jobs.py:1445} INFO - Executor reports meltano_gitlab-to-postgres.extract_load execution_date=2020-07-06 19:00:00+00:00 as failed for try_number 1
[2020-07-06 16:47:30,046] {jobs.py:1461} ERROR - Executor reports task instance <TaskInstance: meltano_gitlab-to-postgres.extract_load 2020-07-06 19:00:00+00:00 [queued]> finished (failed) although the task says its queued. Was the task killed externally?
[2020-07-06 16:47:30,046] {models.py:273} INFO - Filling up the DagBag from /Users/nehiljain/snaptravel/universal-elt-pipeline/airflow-elt/orchestrate/dags/meltano.py
this is a strange error
d
😕
I'm not sure what to make it of. The only Airflow config Meltano sets is listed here: https://gitlab.com/meltano/meltano/blob/master/src/meltano/core/bundle/discovery.yml#L843-857, and none of it relates to the task runner
n
airflow test
is a way to trigger the task manually and it looks like that works correctly. I will dig deeper in meltano.py to see what dag config is being set
d
@nehil_jain I wonder what your 
.meltano/orchestrators/airflow/airflow.cfg
says for 
task_runner
n
good question
d
@nehil_jain Is it possible that config was generated for a newer version of Airflow than the one you've got installed inside Meltano?
Since 
StandardTaskRunner
is defined in the latest Airflow: https://github.com/apache/airflow/blob/master/airflow/task/task_runner/__init__.py#L30
But Meltano is still on 1.10.2
If you delete that 
airflow.cfg
file, it should be regenerated automatically (using the correct version of Airflow) based on that Airflow config in discovery.yml, and anything you've overridden in your meltano.yml 🙂
n
task_runner = StandardTaskRunner
i see
so i can just delete it and run airflow scheduler again ?
d
Yep!
And then you should see it say 
BashTaskRunner
, not 
StandardTaskRunner
19 Views
Open in Slack
PreviousNext
Powered by