{"description":"实验创建于2021/11/24","graph":{"edges":[{"to_node_id":"-1298:input_1","from_node_id":"-1242:data"},{"to_node_id":"-1250:input_data","from_node_id":"-1298:data_1"}],"nodes":[{"node_id":"-1242","module_id":"BigQuantSpace.datahub_load_file.datahub_load_file-v2","parameters":[{"name":"file_path","value":"# Python 动态生成文件路径\ndef bigquant_run():\n # 示例代码如下。在这里编写您的代码\n # 如果不需要动态生成路径,直接返回path即可\n # import datetime\n # import os\n # base_path = \"/var/app/data/datahub\"\n # data_path = os.path.join(base_path, \"bigcrawler\")\n # date = datetime.datetime.now().strftime(\"%Y%m%d\")\n # file_path = os.path.join(data_path, \"{}.csv\".format(date))\n # return file_path \n file_path = './data.csv'\n\n return file_path\n","type":"Literal","bound_global_parameter":null},{"name":"file_type","value":"csv","type":"Literal","bound_global_parameter":null},{"name":"csv_delimiter","value":",","type":"Literal","bound_global_parameter":null},{"name":"other_parameters","value":"{}","type":"Literal","bound_global_parameter":null},{"name":"h5_data_key","value":"data","type":"Literal","bound_global_parameter":null}],"input_ports":[],"output_ports":[{"name":"data","node_id":"-1242"}],"cacheable":false,"seq_num":2,"comment":"","comment_collapsed":true},{"node_id":"-1250","module_id":"BigQuantSpace.datahub_update_datasource.datahub_update_datasource-v5","parameters":[{"name":"alias","value":"my_data","type":"Literal","bound_global_parameter":null},{"name":"primary_key","value":"# #号开始的表示注释，注释需单独一行\n# 主键字段, 每个字段为一行\n# 主键字段主要用于数据去重\ndate\ninstrument\n","type":"Literal","bound_global_parameter":null},{"name":"date_field","value":"date","type":"Literal","bound_global_parameter":null},{"name":"partition_date","value":"无","type":"Literal","bound_global_parameter":null},{"name":"rewrite","value":"False","type":"Literal","bound_global_parameter":null},{"name":"public","value":"True","type":"Literal","bound_global_parameter":null},{"name":"friendly_name","value":"我的数据","type":"Literal","bound_global_parameter":null},{"name":"desc","value":"期货数据","type":"Literal","bound_global_parameter":null},{"name":"only_desc_fields","value":"False","type":"Literal","bound_global_parameter":null},{"name":"fields","value":"\n# #号开始的表示注释，注释需单独一行\n# 对您数据中的每个字段进行描述，系统会根据您定义的数据类型自动将对应列的数据转换\n# 如果是已有数据表，默认会获取已有表的所有数据字段信息\n# eg: {'date': {'desc': '描述', 'type': '数据类型'}, ...}\n# {'instrument': {'desc': '证券代码', 'type': 'str'}, ... }\n{}\n","type":"Literal","bound_global_parameter":null},{"name":"show_doc","value":"True","type":"Literal","bound_global_parameter":null}],"input_ports":[{"name":"input_data","node_id":"-1250"}],"output_ports":[{"name":"data","node_id":"-1250"}],"cacheable":false,"seq_num":1,"comment":"","comment_collapsed":true},{"node_id":"-1298","module_id":"BigQuantSpace.cached.cached-v3","parameters":[{"name":"run","value":"# Python 代码入口函数，input_1/2/3 对应三个输入端，data_1/2/3 对应三个输出端\ndef bigquant_run(input_1, input_2, input_3):\n # 示例代码如下。在这里编写您的代码\n df = input_1.read()\n df.drop(columns='Unnamed: 0',inplace=True)\n df['date'] = df['date'].astype('datetime64')\n data_1 = DataSource.write_df(df)\n \n return Outputs(data_1=data_1, data_2=None, data_3=None)\n","type":"Literal","bound_global_parameter":null},{"name":"post_run","value":"# 后处理函数，可选。输入是主函数的输出，可以在这里对数据做处理，或者返回更友好的outputs数据格式。此函数输出不会被缓存。\ndef bigquant_run(outputs):\n return outputs\n","type":"Literal","bound_global_parameter":null},{"name":"input_ports","value":"","type":"Literal","bound_global_parameter":null},{"name":"params","value":"{}","type":"Literal","bound_global_parameter":null},{"name":"output_ports","value":"data_1","type":"Literal","bound_global_parameter":null}],"input_ports":[{"name":"input_1","node_id":"-1298"},{"name":"input_2","node_id":"-1298"},{"name":"input_3","node_id":"-1298"}],"output_ports":[{"name":"data_1","node_id":"-1298"},{"name":"data_2","node_id":"-1298"},{"name":"data_3","node_id":"-1298"}],"cacheable":true,"seq_num":3,"comment":"","comment_collapsed":true},{"node_id":"-305","module_id":"BigQuantSpace.datahub_bash_operator.datahub_bash_operator-v3","parameters":[{"name":"task_name","value":"当前文件名","type":"Literal","bound_global_parameter":null},{"name":"schedule_interval","value":"0 8 * * *","type":"Literal","bound_global_parameter":null},{"name":"run_now","value":"True","type":"Literal","bound_global_parameter":null}],"input_ports":[],"output_ports":[{"name":"task","node_id":"-305"}],"cacheable":false,"seq_num":4,"comment":"","comment_collapsed":true}],"node_layout":"<node_postions><node_position Node='-1242' Position='351.6296691894531,357.704345703125,200,200'/><node_position Node='-1250' Position='336.3517761230469,531.9898681640625,200,200'/><node_position Node='-1298' Position='354.13873291015625,448.3880615234375,200,200'/><node_position Node='-305' Position='68.72172546386719,315.2794189453125,200,200'/></node_postions>"},"nodes_readonly":false,"studio_version":"v2"}

# 本代码由可视化策略环境自动生成 2021年12月15日 21:24
# 本代码单元只能在可视化模式下编辑。您也可以拷贝代码，粘贴到新建的代码单元或者策略，然后修改。


# Python 动态生成文件路径
def m2_file_path_bigquant_run():
    # 示例代码如下。在这里编写您的代码
    # 如果不需要动态生成路径,直接返回path即可
    # import datetime
    # import os
    # base_path = "/var/app/data/datahub"
    # data_path = os.path.join(base_path, "bigcrawler")
    # date = datetime.datetime.now().strftime("%Y%m%d")
    # file_path = os.path.join(data_path, "{}.csv".format(date))
    # return file_path 
    file_path = './data.csv'

    return file_path

# Python 代码入口函数，input_1/2/3 对应三个输入端，data_1/2/3 对应三个输出端
def m3_run_bigquant_run(input_1, input_2, input_3):
    # 示例代码如下。在这里编写您的代码
    df = input_1.read()
    df.drop(columns='Unnamed: 0',inplace=True)
    df['date'] = df['date'].astype('datetime64')
    data_1 = DataSource.write_df(df)
    
    return Outputs(data_1=data_1, data_2=None, data_3=None)

# 后处理函数，可选。输入是主函数的输出，可以在这里对数据做处理，或者返回更友好的outputs数据格式。此函数输出不会被缓存。
def m3_post_run_bigquant_run(outputs):
    return outputs


m2 = M.datahub_load_file.v2(
    file_path=m2_file_path_bigquant_run,
    file_type='csv',
    csv_delimiter=',',
    other_parameters='{}',
    h5_data_key='data'
)

m3 = M.cached.v3(
    input_1=m2.data,
    run=m3_run_bigquant_run,
    post_run=m3_post_run_bigquant_run,
    input_ports='',
    params='{}',
    output_ports='data_1'
)

m1 = M.datahub_update_datasource.v5(
    input_data=m3.data_1,
    alias='my_data',
    primary_key="""# #号开始的表示注释，注释需单独一行
# 主键字段, 每个字段为一行
# 主键字段主要用于数据去重
date
instrument
""",
    date_field='date',
    partition_date='无',
    rewrite=False,
    public=True,
    friendly_name='我的数据',
    desc='期货数据',
    only_desc_fields=False,
    fields=
# #号开始的表示注释，注释需单独一行
# 对您数据中的每个字段进行描述，系统会根据您定义的数据类型自动将对应列的数据转换
# 如果是已有数据表，默认会获取已有表的所有数据字段信息
# eg: {'date': {'desc': '描述', 'type': '数据类型'}, ...}
# {'instrument': {'desc': '证券代码', 'type': 'str'}, ... }
{}
,
    show_doc=True
)

m4 = M.datahub_bash_operator.v3(
    task_name='当前文件名',
    schedule_interval='0 8 * * *',
    run_now=True
)

[2021-12-15 20:56:24.352963] INFO: 读取数据(文件): 读取数据文件: ./data.csv

[2021-12-15 20:56:24.429718] INFO: moduleinvoker: datahub_load_file.v2 运行完成[0.169342s].

[2021-12-15 20:56:24.452804] INFO: moduleinvoker: cached.v3 开始运行..

[2021-12-15 20:56:24.604543] INFO: moduleinvoker: cached.v3 运行完成[0.151777s].

[2021-12-15 20:56:24.666336] WARNING: 更新入库: 您选择了覆盖原数据，这将会把数据库中 my_data 的旧数据覆盖，请谨慎操作！(系统将在提示3次后执行您的操作)

[2021-12-15 20:56:27.669915] WARNING: 更新入库: 您选择了覆盖原数据，这将会把数据库中 my_data 的旧数据覆盖，请谨慎操作！(系统将在提示3次后执行您的操作)

[2021-12-15 20:56:30.674911] WARNING: 更新入库: 您选择了覆盖原数据，这将会把数据库中 my_data 的旧数据覆盖，请谨慎操作！(系统将在提示3次后执行您的操作)

[2021-12-15 20:56:33.859351] INFO: 更新入库: 正在写入数据，请稍后... 

[2021-12-15 20:56:33.863994] INFO: bigdatasource: start remote_update  my_data ... 

[2021-12-15 20:56:34.604643] INFO: bigdatasource: remote update success!

[2021-12-15 20:56:34.607137] INFO: 更新入库: 数据更新完成, 可以使用  DataSource('my_data').read()  来查看数据!

[2021-12-15 20:56:34.609755] INFO: moduleinvoker: datahub_update_datasource.v5 运行完成[9.978093s].