zoukankan      html  css  js  c++  java
  • How to implement connection pool in spark streaming

    在spark streaming的文档里,有这么一段:

    def sendPartition(iter):
        # ConnectionPool is a static, lazily initialized pool of connections
        connection = ConnectionPool.getConnection()
        for record in iter:
            connection.send(record)
        # return to the pool for future reuse
        ConnectionPool.returnConnection(connection)
    
    dstream.foreachRDD(lambda rdd: rdd.foreachPartition(sendPartition))

    但是怎么让worker得到一个ConectionPool呢?简单的想法是在使用static变量指向一个ConnectionPool。但这里有一个讲究:怎么保证这个ConnectionPool是worker上的,而不是driver上的?

    用pyhton为例:

    在ConnectionPool.py里实现一个pool

    #/usr/bin/python
    #connection_pool.py
    import
    psycopg2 import settings from DBUtils.PooledDB import PooledDB pool = PooledDB(psycopg2, settings.connection_pool_size, host=settings.db_host, database=settings.database, user=settings.db_user, password=settings.db_password)
    def getConnection():
    return pool.connection()

    假设stream的主代码在main.py里,提交spark

    spark-submit --py-files connection_pool.py main.py

    这样connection_pool.py将被发送到worker执行,main.py里的 sendPartition 在worker节点上执行的时候就可以获得ConnectionPool.getConnection()调用。

    这里的关键是明白哪些代码在driver上跑,哪些在worker上跑。

     
  • 相关阅读:
    element_2对话框
    填报
    润乾报表中进度条的一种实现方式
    列名作为分类值时如何画出统计图
    填报之动态扩展列
    自由格式填报的制作
    复杂报表设计之动态报表
    如何通过动态参数实现周报制作
    如何实现行列互换效果?
    大数据集报表点击表头排序
  • 原文地址:https://www.cnblogs.com/englefly/p/4579863.html
Copyright © 2011-2022 走看看