git.net

[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]

Re: Beam python sdk with gevent


Thanks Ahmet for providing a reference code. I will give it a try.

I also tried to read the code it feels like you are using the multiprocess for parallelizing runtime jobs. We wanted to use Gevent because it is lightweight and good for parallelizing IO/Network bound jobs. I would also recommend providing Gevent support in the future because it can efficiently use resources and it can scale well in heavy load.



On Fri, Sep 21, 2018 at 5:58 PM Ahmet Altay <altay@xxxxxxxxxx> wrote:
Thank you for the example it helps.

I still do not know what is wrong with gevent. Would you consider using multiprocessing package? We are already using that to accomplish something similar in file based sinks, and there is already utility function that wraps it around similar to your example [1].


Ahmet


On Wed, Sep 19, 2018 at 10:25 PM, Rakesh Kumar <rakeshkumar@xxxxxxxx> wrote:

Gevent is basically used to make parallel network calls. We are using gevent in one of the transformation methods to call internal services. The transformation method is making multiple network call in parallel. Here is the code snippet:
/__init__.py
import gevent.monkey
gevent.monkey.patch_all()
 
/transform.py
from gevent import Greenlet
from gevent import joinall
def filter_out_invalid_users(events):
   key, user_id_data_pairs = events
   user_ids = [user_id for user_id, data in user_id_data_pairs]

   jobs = []
   id_chunks = utils.chunk_list_evenly(user_ids, BATCH_SIZE)
   for id_chunk in id_chunks:
      jobs.append(Greenlet.spawn(_call_users_service, # _call_user_service_ method is making the network call.
                                 list(id_chunk)))

   """
   Here we increase the timeout based on the number of greenlets we are running, to account for yielding
   among greenlets
   """
   join_timeout = GREENLET_TIMEOUT + len(jobs) * GREENLET_TIMEOUT * 0.1
   joinall(jobs, timeout=join_timeout)

   successful_jobs = [job for job in jobs if job.successful()]
   valid_user_ids = []
   for job in successful_jobs:
      network_response = job.get()
      valid_user_ids.append(network_response.user_id)
   yield valid_user_ids

def _call_users_service(user_ids):
   # make network call and return response
   ..
   ..
   return network_response

On Tue, Sep 18, 2018 at 7:07 PM Ahmet Altay <altay@xxxxxxxxxx> wrote:
I am also not familiar with gevent. Could you explain what are you trying to do and how do you plan to use gevent?

On Tue, Sep 18, 2018 at 9:38 AM, Lukasz Cwik <lcwik@xxxxxxxxxx> wrote:
I don't think anyone has tried what your doing. The code that your working with is very new.

On Mon, Sep 17, 2018 at 5:02 PM Micah Wylde <mwylde@xxxxxxxx> wrote:
Hi all,

We're using the Python SDK with the portable Flink runner and running into some problems integrating gevent. We're patching the gRPC runtime for gevent as described in [0] which allows pipelines to start and partially run. However the tasks produce a stream of gevent exceptions:

Exception greenlet.error: error('cannot switch to a different thread',) in 'grpc._cython.cygrpc.run_loop' ignored
Traceback (most recent call last):
  File "src/gevent/event.py", line 240, in gevent._event.Event.wait
  File "src/gevent/event.py", line 140, in gevent._event._AbstractLinkable._wait
  File "src/gevent/event.py", line 117, in gevent._event._AbstractLinkable._wait_core
  File "src/gevent/event.py", line 119, in gevent._event._AbstractLinkable._wait_core
  File "src/gevent/_greenlet_primitives.py", line 59, in gevent.__greenlet_primitives.SwitchOutGreenletWithLoop.switch
  File "src/gevent/_greenlet_primitives.py", line 59, in gevent.__greenlet_primitives.SwitchOutGreenletWithLoop.switch
  File "src/gevent/_greenlet_primitives.py", line 63, in gevent.__greenlet_primitives.SwitchOutGreenletWithLoop.switch
  File "src/gevent/__greenlet_primitives.pxd", line 35, in gevent.__greenlet_primitives._greenlet_switch
greenlet.error: cannot switch to a different thread

and do not make any progress.

Has anybody else successfully used the portable python sdk with gevent? Or is there a recommended alternative for doing async IO in python pipelines?


Micah

--
Rakesh Kumar
Software Engineer  

--
Rakesh Kumar
Software Engineer  
510-761-1364 |