小编CCl*_*rke的帖子

我有一个数据库表,其中一列包含不同的17位数字.当我要求R将该表的一个部分或全部读入数据帧时,它会复制一些值4或5次并删除其他值.结果是,具有17个唯一值的表将以6结尾.这很奇怪.当我使用RODBC连接读取数据或者是否将它们保存到csv并以这种方式加载它时,就会发生这种情况.这绝对是一个R问题,因为我已经用假数据隔离了这个问题:

base_num <- "9600000005206"

# Now create a vector of unique numbers in character format
x <- paste0(base_num, seq(5680, 5760, 1))

x

x
[1] "96000000052065680" "96000000052065681" "96000000052065682" 
"96000000052065683" "96000000052065684" "96000000052065685" 
"96000000052065686"
[8] "96000000052065687" "96000000052065688" "96000000052065689" 
"96000000052065690" "96000000052065691" "96000000052065692" 
"96000000052065693"
[15] "96000000052065694" "96000000052065695" "96000000052065696" 
"96000000052065697" "96000000052065698" "96000000052065699" 
"96000000052065700"

# Convert them to numeric
# ta-da! Duplicate values and dropped values
as.numeric(x)

 [1] 96000000052065680 96000000052065680 96000000052065680 
 96000000052065680 96000000052065680 96000000052065680 96000000052065680
 [8] 96000000052065680 96000000052065696 96000000052065696 
 96000000052065696 96000000052065696 96000000052065696 96000000052065696
 [15] 96000000052065696 96000000052065696 96000000052065696 …

我刚刚将 Airflow 升级到 2.2.5，但无法使用 EmptyOperator。它应该很简单from airflow.operators.empty import EmptyOperator，但我收到错误ModuleNotFoundError: No module named 'airflow.operators.empty'。我也尝试过：

from airflow.operators import empty
from empty.operators import EmptyOperator

Airflow 存储库本身显示了应该可以工作的结构 from airflow.operators.empty import EmptyOperator，但事实并非如此，所以我对发生的事情感到非常困惑。

编辑：我已经看到了 SA 上关于此问题的所有问题，它们都给了我我在这里询问的错误 - 请您将其保持打开状态以便我可以获得一些帮助吗？

我有一个可以使用 Bash 非常简单地读取的文件，如下所示： gzip -d -c my_file.json.gz | jq . 这确认它是有效的 JSON。但是当我尝试使用 Python 来读取它时，如下所示：

import json
import gzip
with gzip.open('my_file.json.gz') as f:
    data = f.read() # returns a byte string `b'`
json.loads(data)

我收到错误：

json.decoder.JSONDecodeError: Extra data: line 2 column 1 (char 1632)

但我知道它是来自我的 Bash 命令的有效 JSON。我已经被这个看似简单的问题困扰了很长一段时间，并尝试了一切感觉。有人可以帮忙吗？谢谢。

我有几个可以同时运行的任务。当他们完成后，我需要运行最后一个任务。我尝试使用任务分组来做到这一点，如下所示：

import airflow
from airflow.utils.task_group import TaskGroup

with airflow.DAG(
        'my_dag',
        catchup=False,
        default_args=default_args,
        schedule_interval=datetime.timedelta(days=1),
    ) as dag:

    with TaskGroup(group_id='task_group_1') as tg1:

    
        task1 = MyOperator(
            task_id='task1',
            dag=dag,
        )

        task2 = MyOperator(
            task_id='task2',
            dag=dag,
        )
        
        [task1, task2]    
    
    final_task = MyOtherOperator(
        task_id="final_task",
        dag=dag
    )

    tg1 >> final_task
   

然而，这里发生的情况是 Final_task 在任务组中的每个任务之后运行多次，因此：

任务1 -> 最终任务 任务2 -> 最终任务

我想要的是任务组并行运行，并且当最终任务完成时只运行一次，如下所示：

[任务1，任务2] -> 最终任务

我认为使用任务组可以帮助我完成此要求，但它没有按预期工作。有人可以帮忙吗？谢谢。

编辑：这是 Airflow 文档示例的结果。它导致task3在group.task1和group1.task2之后运行。我需要它在两个分组任务完成后只运行一次。

最后编辑：事实证明我误解了树视图 - 图形视图确认了分组操作，尽管我仍然收到最终任务的一些其他错误。感谢您帮助我了解有关 DAG 的更多信息。