python - 气流 2 - ModuleNotFoundError:没有名为“airflow.operators.text_processing_plugin”的模块
问题描述
我是新手,airflow
并试图为处理文本做一个 dag。我有一个由文本处理任务组成的数据管道——读取文档、清理文本并将数据加载到 JSON 文件。对于文本处理,自定义运算符用于每个转换任务,它们保存在text_processing_plugin
文件夹中。文件夹的完整文件夹结构plugin
是:-
├── airflow.cfg
├── airflow.db
├── airflow-webserver.pid
├── dags
│ ├── d0.py
├── plugins
│ └── text_processing_plugin
│ ├── __init__.py
│ ├── operators
│ │ ├── dependency_parsing.py
│ │ ├── entity_detection.py
│ │ ├── __init__.py
│ │ ├── lemmatize.py
│ │ ├── pos_tagging.py
│ │ ├── remove_stop_words.py
│ │ └── tokenize_doc.pyfolder structure of plugin folder is:-
├── requirements.txt
├── unittests.cfg
哪里text_processing_plugin/__init__.py
有以下代码: -
from airflow.plugins_manager import AirflowPlugin
from text_processing_plugin.operators.dependency_parsing import DependencyParsingOperator
from text_processing_plugin.operators.entity_detection import DetectEntityOperator
from text_processing_plugin.operators.lemmatize import LemmatizeOperator
from text_processing_plugin.operators.pos_tagging import POSTagOperator
from text_processing_plugin.operators.remove_stop_words import RemoveStopWordsOperator
from text_processing_plugin.operators.tokenize_doc import DocTokenizerOperator
class TextProcessingPlugin(AirflowPlugin):
name = "text_processing_plugin"
operators = [DependencyParsingOperator, DetectEntityOperator, LemmatizeOperator, POSTagOperator,
RemoveStopWordsOperator, DocTokenizerOperator]
sensors = []
hooks = []
executors = []
macros = []
admin_views = []
flask_blueprints = []
menu_links = []
appbuilder_views = []
appbuilder_menu_items = []
global_operator_extra_links = []
operator_extra_links = []
为了制作 DAG,airflow 1.x
就像下面使用的范例:-
import os
import json
import spacy
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
from airflow.operators.text_processing_plugin import DependencyParsingOperator, DetectEntityOperator, LemmatizeOperator, POSTagOperator, RemoveStopWordsOperator, DocTokenizerOperator
sp = spacy.load('en_core_web_sm')
default_args = {
'owner': 'episource',
'depends_on_past': True,
'start_date': datetime.datetime(2021, 3, 30),
'retries': 0,
'schedule_interval':'@once',
}
dag = DAG(
'text_processing_dag',
description='Text Processing Dag',
default_args=default_args,
catchup=False,
tags=['text_processing'])
def read_doc(**kwargs):
file_path = os.path.join(os.getcwd(), '/data/1.txt')
doc = open(file_path).read()
return doc
def write_to_json(**kwargs):
ti = kwargs['ti']
with open(os.path.join(os.getcwd, 'output', '1.json'), 'a+') as file:
result_1 = ti.xcom_pull(task_ids = 'tokenize_doc')
result_2 = ti.xcom_pull(task_ids = 'detect_entity')
print('result 1 is ', result_1)
print('result 2 is ', result_2)
file.write(json.dumps(result_1))
file.write(json.dumps(result_2))
extract = PythonOperator(
task_id = 'extract',
python_callable = read_doc,
dag = dag)
t11_tokenize_doc = DocTokenizerOperator(
sp = sp,
task_id = "transform_tokenize_doc",
dag = dag,
name = "Sentence Tokenizing",
pool='t1',
task_concurrency=2)
t12_detect_entities = DetectEntityOperator(
sp = sp,
task_id = "transform_detect_entity",
dag = dag,
name = "Entity Detection",
pool='t1',
task_concurrency=2)
load = PythonOperator(
task_id = 'load',
python_callable = write_to_json,
dag = dag)
extract >> [t11_tokenize_doc, t12_detect_entities] >> load
当我尝试运行代码时,我得到:-
Traceback (most recent call last):
File "dags/d0.py", line 8, in <module>
from airflow.operators.text_processing_plugin import DependencyParsingOperator, DetectEntityOperator, LemmatizeOperator, POSTagOperator, RemoveStopWordsOperator, DocTokenizerOperator
ModuleNotFoundError: No module named 'airflow.operators.text_processing_plugin'
我在 Stackoverflow 上引用了一些现有的答案,但无法解决该错误。将不胜感激一些提示。
解决方案
在2.0版更改:不再支持通过气流导入插件中添加的运算符、传感器、钩子。{operators,sensors,hooks}.<plugin_name> ,这些扩展应该作为常规 python 模块导入。有关更多信息,请参阅:模块管理 和创建自定义 Operator
推荐阅读
- postgresql - Postgresql 按列分区,没有主键
- http-live-streaming - Video.js HLS 选择质量低于预期的 TS 片段
- sql - @timestamp 的 Elasticsearch SQL 过滤不起作用
- python - Pip 安装 32 位包而不是 64 位
- java - 如何解决 Java 中的 switch case 错误?
- python - 无法导入 spacy
- javascript - 在 javascript 中以 200 毫秒的时间步长插值坐标
- ansible - 从库存中选择一台可用的服务器
- java - 列表到列表副本之间的Java差异
- pickle - 如何从 .pkl 文件加载预训练 GAN?