Function sub_process

union/D3L/opendata_large_query.py:33–82

(query_tables, queue)

    return result

def sub_process(query_tables, queue):
    tables_to_skip = ["USA_CSV0000000000001722"]
    for i, table_name_with_extension in enumerate(query_tables):
        
        table_name = os.path.splitext(table_name_with_extension)[0]
        if table_name in tables_to_skip:
            print(f"跳过查询表 {i + 1}，因为它在要跳过的列表中")
            queue.put(1)  # 在队列中放入一个占位符值
            continue
        output_folder = "/home/wangyanzhang/d3l-main/d3l-main/examples/notebooks/opendata_large_50"
        if not os.path.exists(output_folder):
            os.makedirs(output_folder)
        results_file = os.path.join(output_folder, f"{table_name}.csv")

        if os.path.exists(results_file):
            print(f"跳过查询表 {i + 1}，因为结果文件已经存在：{results_file}")
            queue.put(1)  # 在队列中放入一个占位符值
            continue

        try:
            # 执行查询
            results, extended_results = qe.table_query(table=dataloader.read_table(table_name=table_name),
                                                aggregator=None, k=50, verbose=True)


            # 创建一个新的 CSV 文件
            with open(results_file, mode='w', newline='') as csvfile:
                writer = csv.writer(csvfile)

                # Write the header
                writer.writerow(['query_table', 'candidate_table', 'query_col_name', 'candidate_col_name'])

                # 写入查询结果到 CSV 文件
                for result in extended_results:
                    query_table = table_name
                    candidate_table = os.path.basename(result[0])

                    for x, column_info in enumerate(result[1]):
                        column_info_name = f"column_info{x+1}"
                        globals()[column_info_name] = column_info[0]

                        row = [f"{query_table}.csv", f"{candidate_table}.csv", column_info[0][0], column_info[0][1]]
                        writer.writerow(row)

            print(f"Results for query table {i + 1} have been written to {results_file}")
            queue.put(1)
        except Exception as e:
            # 在这里处理异常。你可以打印错误消息以进行调试。
            print(f"查询表 {i + 1} 发生了异常：{e}")
    queue.put((-1, "test-pid"))

if __name__ == "__main__":
    # 记录程序开始时间