0

我一直在使用以下脚本上传数据来对我的模块进行负载测试：

import json
import ast
import pandas as pd
import sys
import cloudant_connection as cloud

df = pd.read_csv("./deviceId_data/device_starts_"+ sys.argv[1] + ".csv")
print(" checkpoint 1 cleared ")
def push_data_to_cloudant(ID,Key,Value,Database):
    Value = ast.literal_eval(Value)
    temp_doc = {}
    temp_doc["_id"] = ID
    temp_doc["value"] = Value["value"]
    temp_doc["devId"] = Value["devId"]
    temp_doc["eDateTime"] = Key[0]
    temp_doc["eDate"] = Value["eDate"]
    temp_doc["cDateTime"] = Key[0]
    temp_doc["cDate"] = Value["cDate"]
    
    new_doc = Database.create_document(temp_doc)
    
    if new_doc.exists():
        #print("doc created")
        return "Success"
    else:
        print("Failed in pushing document")
        return "Failure"

with open("./connection_config_source.json") as f:
    connect_conf = json.load(f)
print(" checkpoint 2 cleared ")
API_KEY = connect_conf['cloudant_api_key']
ACC_NAME = connect_conf['cloudant_account_name']
print(" checkpoint 3 cleared ")
try:
    client = cloud.connecting_to_cloudant_via_api(ACC_NAME,API_KEY)

    database_name = 'DB_NAME'
    
    Database = client[database_name]
    print(" checkpoint 4 cleared ")
    if Database.exists():
        print("Connected")
      
    status = [push_data_to_cloudant(ID,Key,Value,Database) for (ID,Key,Value) in zip(df['id'],df['key'],df['value'])]
    print(" last checkpoint cleared ")
except Exception as e:
    print("Failed:" + str(e))

我知道有比使用列表理解更快的方法。但我不知道如何在这种情况下使用它们。

我知道 df.apply() 比这更快，但我想知道我是否可以在这个用例中使用 Pandas Vectorization 或 Numpy Vectorization。

4

1 回答 1

0

python-cloudant 文档：

bulk_docs(docs) 通过单个请求执行多个文档插入和/或更新。每个文档必须是或扩展一个字典，就像 Document 和 DesignDocument 对象一样。如果要更新文档，则文档必须包含 _id 和 _rev 字段。

参数： docs (list) – 要创建/更新的文档列表。返回：JSON 格式的批量文档创建/更新状态

只需使用：

Database = client['DB_name']

Database.bulk_docs(*argv)

这里的参数可以是字典列表或 json 对象。

于 2022-01-12T07:04:41.977 回答