如何编写代码将日志记录为 argv？

Question

在终端中，如果我执行

docker run -it --rm --name consumer --link zookeeper:zookeeper --link kafka:kafka       \
debezium/kafka:1.1 watch-topic -a bankserver1.bank.holding > C:/Users/User/python_test  \
/holding_pivot.txt

日志结果会像这样写入文件holding_pivot.txt

WARNING: Using default BROKER_ID=1, which is valid only for non-clustered installations.
Using ZOOKEEPER_CONNECT=172.17.0.3:2181
Using KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://172.17.0.6:9092
Using KAFKA_BROKER=172.17.0.4:9092
Contents of topic bankserver1.bank.holding:
{"schema":{"type":"struct","fields":[{"type":"struct","fiel...
{"schema":{"type":"struct","fields":[{"type":"struct","file...

我想要一个 python 文件在写入文件日志之前处理它（我不想要一个从日志中读取的 python 文件（因为我必须区分哪一行已经被读取和处理，哪一行没有））

代码的过程是这样的

RESULT -> Python 文件（句柄）-> 写入日志文件。

主要问题：哪个 Python 库支持它？你能给我一个示例代码吗？

我看到一个这样的文件：

docker run -it --rm --name consumer --link zookeeper:zookeeper --link kafka:kafka \
debezium/kafka:1.1 watch-topic -a bankserver1.bank.holding \
| grep --line-buffered '^{' | pipe > ./stream.py > ./holding_pivot.txt

====================== 更新======================= 根据@piertoni，我添加了这个

docker run -it --rm --name consumer --link zookeeper:zookeeper --link /
kafka:kafka  debezium/kafka:1.1 watch-topic -a bankserver1.bank.holding /
| python C:/Users/User/python_test/test_pipe.py > C:/Users/User/python_test /
/holding_pivot.txt

这是我的尝试test_pipe.py

#!/usr/bin/env python3 -u
# Note: the -u denotes unbuffered (i.e output straing to stdout without buffering data and then writing to stdout)

#!/usr/bin/env python3
import fileinput
import json
import os
import sys
from datetime import datetime

for line in sys.stdin:
    with open('log.txt', 'a') as wr:
        wr.write("Pipe success")

with open('log.txt', 'a') as wr:
    wr.write("Pipe success")

with fileinput.input() as f:
    for line in f:
        with open('log.txt', 'a') as wr:
            wr.write(f"Argument List: {str(line)}")

但是，log.txt 文件中没有写入任何内容（我只需要 python 文件可以读取初始管道的结果）。

Answer 1

这确实与linked question 完全相同，但为了使事情完全明确，以下是使用 python 编写流过滤器的方法：

您需要从标准输入读取，处理然后写入标准输出。前者可以通过多种方式完成，但最简单的是使用fileinput。后者也可以通过多种方式完成，但如果需要，我个人只会使用print 和明确的sys.stdout.flush()。因此将它们放在一起：

import fileinput
import sys

with fileinput.input() as f:
    for line in f:
        line = line.title() # filter your lines here
        print(line, end="")
        sys.stdout.flush()

dir | python test.py # try this first, to check it works
docker_cmd | python test.py # try this second
docker_cmd | python test.py > log.txt # send the output to the log.

请注意，要么您将标准输出发送到带有 shell 重定向的文件或您在 python 中处理文件写入（除非您计划将不同的内容打印到标准输出而不是您编写的到文件，这似乎不是这种情况）。

缓冲

缓冲可能意味着两件事：终端缓冲流和python自己的缓冲。 -u（至少在 python 3.9 中）不会取消缓冲标准输入。 this question 中讨论了一些取消缓冲的技术。使用 strace 运行确认您确实可以取消缓冲输入（至少在我的系统上）所以这样的事情应该避免所有 python 的 缓冲：

import sys

while line x:= sys.stdin.buffer.raw.readline():
    print("line".decode())

这在我的系统上比迭代标准输入要慢，大概是因为decode。你可以试试：

import os
u_stdin = os.fopen(sys.stdin.fileno(), "r", buffering=1)

为您提供行缓冲的文本标准输入流。显然，破坏迭代标准输入对象的bug with .next 已得到修复。但是，如果 terminal 正在缓冲管道，那么这一切都不是好事。在 linux 上我可以unbuffer。 This question 提出了一种方法，可能让事情为你工作。

如果您能够提供可重现的缓冲命令，我可以进一步测试。我可以使用man ssh | 等（或tail -f，同时附加到文件）获得缓冲流，但无法重现读取失败。

最后，我假设你的 docker cmd 真的输出到 stdout？不是，比如说，stderr？如果在 docker 命令中添加 2>&1 以将 stderr 发送到 stdout，会发生什么情况？你能设置一个在几秒钟内运行完成的 docker 命令，这样你就可以测试 python 从不读取任何内容？缓冲应该会减慢读取速度，但它应该不阻止数据通过。