Function sub_process

union/D3L/opendata_small_query.py:33–74

(query_tables, queue)

    return result

def sub_process(query_tables, queue):
    for i, table_name_with_extension in enumerate(query_tables):

        table_name = os.path.splitext(table_name_with_extension)[0]
        output_folder = "/home/wangyanzhang/d3l-main/d3l-main/examples/notebooks/opendata_large_60"
        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

        
        # 执行查询
        results, extended_results = qe.table_query(table=dataloader.read_table(table_name=table_name),
                                                aggregator=None, k=60, 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)
    queue.put((-1, "test-pid"))

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