python - 带有 HappyBase 连接池的 PySpark dataframe.foreach() 返回 'TypeError: can' t pickle thread.lock 对象'

Question

我有一个 PySpark 作业可以更新 HBase 中的一些对象(Spark v1.6.0；happybase v0.9)。

如果我为每一行打开/关闭一个 HBase 连接，它会有点工作:

def process_row(row):
    conn = happybase.Connection(host=[hbase_master])
    # update HBase record with data from row
    conn.close()

my_dataframe.foreach(process_row)

几千次更新插入后，我们开始看到这样的错误:

TTransportException: Could not connect to [hbase_master]:9090

显然，为每个更新插入打开/关闭连接是低效的。这个函数实际上只是一个正确解决方案的占位符。

然后我尝试创建一个使用连接池的 process_row 函数版本:

pool = happybase.ConnectionPool(size=20, host=[hbase_master])

def process_row(row):
    with pool.connection() as conn:
        # update HBase record with data from row

由于某种原因，此函数的连接池版本返回错误(参见 complete error message ):

TypeError: can't pickle thread.lock objects

你能看出我做错了什么吗？

更新

我看到了this post并怀疑我遇到了同样的问题:Spark 尝试序列化 pool 对象并将其分发给每个执行程序，但不能在多个执行程序之间共享此连接池对象。

听起来我需要将数据集分成多个分区，并为每个分区使用一个连接(参见 design patterns for using foreachrdd)。我根据文档中的示例尝试了此操作:

def persist_to_hbase(dataframe_partition):
    hbase_connection = happybase.Connection(host=[hbase_master])
    for row in dataframe_partition:
        # persist data
    hbase_connection.close()

my_dataframe.foreachPartition(lambda dataframe_partition: persist_to_hbase(dataframe_partition))

不幸的是，它仍然返回“无法腌制 thread.lock 对象”错误。

Answer 1

从根本上说，happybase 连接只是 tcp 连接，因此它们不能在进程之间共享。连接池主要用于多线程应用程序，也证明对单线程应用程序有用，单线程应用程序可以将池用作具有连接重用的全局“连接工厂”，这可以简化代码，因为不需要传递“连接”对象大约。它还使错误恢复更容易一些。

在任何情况下，池(只是一组连接)都不能在进程之间共享。出于这个原因，尝试序列化它没有意义。 (池使用导致序列化失败的锁，但这只是一个症状。)

也许您可以使用一个助手来有条件地创建一个池(或连接)并将其存储为模块局部变量，而不是在导入时实例化它，例如

_pool = None

def get_pool():
    global _pool
    if _pool is None:
        _pool = happybase.ConnectionPool(size=1, host=[hbase_master])
    return pool

def process(...)
    with get_pool().connection() as connection:
        connection.table(...).put(...)

这会在首次使用时而不是在导入时实例化池/连接。