Python InfluxDB2 - write_api.write(...) 如何检查是否成功？

Question

我需要将历史数据写入 InfluxDB（我使用的是 Python，在这种情况下这不是必须的，所以我可能愿意接受非 Python 的解决方案）。我这样设置写API

write_api = client.write_api(write_options=ASYNCHRONOUS)

数据来自一个以时间戳为key的DataFrame，所以我这样写到数据库中

result = write_api.write(bucket=bucket, data_frame_measurement_name=field_key, record=a_data_frame)

此调用不会抛出异常，即使 InfluxDB 服务器已关闭。 result 有一个受保护的属性 _success，它在调试时是一个布尔值，但我无法从代码中访问它。

如何检查写入是否成功？

Answer 1

如果您想立即将数据写入数据库，请使用 write_api 的同步版本 - https://github.com/influxdata/influxdb-client-python/blob/58343322678dd20c642fdf9d0a9b68bc2c09add9/examples/example.py#L12

应该通过调用“触发”异步写入 .get() - https://github.com/influxdata/influxdb-client-python#asynchronous-client

此致

Answer 2

write_api.write() returns a multiprocessing.pool.AsyncResult 或 multiprocessing.pool.AsyncResult（两者相同）。

使用此 return 对象，您可以通过多种方式检查异步请求。看这里：https://docs.python.org/2/library/multiprocessing.html#multiprocessing.pool.AsyncResult

如果可以使用阻塞请求，那么write_api = client.write_api(write_options=SYNCRONOUS)就可以使用。

Answer 3

如果您使用后台批处理，您可以添加自定义成功、错误和重试回调。

from influxdb_client import InfluxDBClient

def success_cb(details, data):
    url, token, org = details
    print(url, token, org)
    data = data.decode('utf-8').split('\n')
    print('Total Rows Inserted:', len(data))  

def error_cb(details, data, exception):
    print(exc)

def retry_cb(details, data, exception):
    print('Retrying because of an exception:', exc)    


with InfluxDBClient(url, token, org) as client:
    with client.write_api(success_callback=success_cb,
                          error_callback=error_cb,
                          retry_callback=retry_cb) as write_api:

        write_api.write(...)

如果你急于测试所有的回调，不想等到所有的重试结束，你可以覆盖重试的间隔和次数。

from influxdb_client import InfluxDBClient, WriteOptions

with InfluxDBClient(url, token, org) as client:
    with client.write_api(success_callback=success_cb,
                          error_callback=error_cb,
                          retry_callback=retry_cb,
                          write_options=WriteOptions(retry_interval=60,
                                                     max_retries=2),
                          ) as write_api:
        ...

Python InfluxDB2 - write_api.write(...) 如何检查是否成功？

Python InfluxDB2 - write_api.write(...) How to check for success?

python

pandas

dataframe

influxdb-python

influxdb-2