for df in dataframes:
    df = df.where(...)
    df = df.with_column(...) 
    df.write_parquet(...)  # each execution here will block and use all available resources on your machine

df = df.concat(df2).concat(df3)...
df = df.where(...)
df = df.with_column(...) 
df.write_parquet(...)

link_data_directed_df = unique_nodes_df[(unique_nodes_df["from_column"].isin(nodes_id_list))

& (unique_nodes_df["to_column"].isin(nodes_id_list))]

pd.read_csv

daft.read_csv

| group_id | node_id | node_data...|

groupby

.map_groups

# Use daft to read data
df = daft.read_parquet(...)
df.collect()

# Create a bounded queue and a pool of workers that continuously pull from the queue
queue = Queue(...)
workers = ...

# For every iteration, put work into the queue
for iteration in iterations:
    pandas_df = df.where(...).to_pandas()
    queue.put(pandas_df)  # block if the queue is full!

@daft.udf (return_dtype=daft.DataType. string())

def get_node_id(from_node_id):

nodes=[]

for node in <http://from_node_id.to|from_node_id.to>_pylist():

nodes.append(community_daft_df.where(community_daft_df ["NODE"] ==                                                             node).collect (). to_pydict() ["COMMUNITY"][0])

return nodes

df = edge_100_sample_df.with_column("community",get_node_id(edge_100_sample_df["FROM_HESAP_NO_MBB_NO"]))

df.collect()

"my_column_name"

| my_column_name: Struct[foo: list[int64], bar: list[int64]] |
| {"foo": [1, 2, 3], "bar": [4, 5, 6]}                       |
| {"foo": [7, 8, 9], "bar": []}                              |

return Series.from_pylist([
    {
        "node_id": G_directed_obj.vs['name'][i],
        "authority": G_directed_obj.authority_score(weights=<http://G_directed_obj.es|G_directed_obj.es>('weight'))[i],
        ...
    } for i in range(community_member_count_int)
])

@daft.udf (return_dtype=daft.DataType.struct(fields=fieldcs_dict))
def centrality_matrix_calculation(influence_df):
    inf1_list = []
    inf2_list = []
    node_id_list = G_directed_obj.vs["name"]
    
    print ("influence calculations")
    
    for node in node_id_list:
        print (node)

        cur_inf_out_data_daft_df = influence_df.where(influence_df["FROM_HESAP_NO_MBB_NO"] == node).collect()
        cur_inf_to_data_daft_df = influence_df.where(influence_dF["TO_HESAP_NO_MBB_NO"] == node).collect()

        print("get influence subdfs")
        
        sum_outdegree_weights_float = cur_inf_out_data_daft_df.sum("EXPRESSION").collect().to_pydict()["EXPRESSION"][0]
        sum_indegree_weights_float = cur_inf_to_data_daft_df.sum("EXPRESSION").collect().to_pydict()["EXPRESSION"][0]

        print ("sum")
        
        try:
            inf1_list.append(sum_outdegree_weights_float / community_member_count_int)
        except TypeError:
            inf1_list.append(0)
        
        print ("append1")
        
        try:
            inf2_list.append(sum_indegree_weights_float / community_member_count_int)
        except TypeError:
            inf2_list.append(0)
        
        print("append2")
    
    centrality_matrix_dict = {}

    centrality_matrix_dict["influence1_val"] = inf1_list
    centrality_matrix_dict["influence2_val"]=inf2_list
    centrality matrix_dict["influence3_val"] = list(map(sum, zip(inf1_list, inf2_list)))

    return daft.Series.from_pylist([matrix_dict])
    

    edge_sample10_df.groupby("COMMUNITY_ID").map_groups(centrality_matrix_calculation(
                                                    edge_sample10_df[["FROM_HESAP_NO_MBB_NO","TO_HESAP_NO_MBB_NO","EXPRESSION"]]
                                        )).collect()

"edge_sample10_df.groupby("COMMUNITY_ID").map_groups(centrality_matrix_calculation"