Python库 | bqueryd-0.3.0.tar.gz

# Bqueryd A companion library to [bquery](https://github.com/visualfabriq/bquery/) to make distributed bquery calls possible. Think of it as a far more rudimentary alternative to [Hadoop](http://hadoop.apache.org/) or [Dask](https://dask.pydata.org/en/latest/) ## The Idea Web applications or client analysis tools do not perform the heavy lifting of calculations over large sets of data themselves, the data is stored on a collection of other servers which respond to queries over the network. Data files that are used in computations are stored in [bcolz](http://bcolz.blosc.org/en/latest/) files. For _really_ large datasets, the bcolz files can also be split up into 'shards' over several servers, and a query can then be performed over several servers and the results combined to the calling function by the bqueryd library. ## Getting started Make sure you have Python virtualenv installed first. As a start we need some interesting data, that is reasonably large in size. Download some Taxi data from the [NYC Taxi & Limousine Commission](http://www.nyc.gov/html/tlc/html/about/trip_record_data.shtml) virtualenv bqueryd_getting_started cd bqueryd_getting_started . bin/activate wget https://s3.amazonaws.com/nyc-tlc/trip+data/yellow_tripdata_2016-01.csv pip install bqueryd pandas We are only downloading the data for one month, a more interesting test is of course download the data for an entire year. But this is a good start. The data for one month is already 10 million records. Run ipython, and let's convert the CSV file to a bcolz file. import bcolz import pandas as pd data = pd.read_csv('yellow_tripdata_2016-01.csv', parse_dates=['tpep_pickup_datetime', 'tpep_dropoff_datetime']) ct = bcolz.ctable.fromdataframe(data, rootdir='tripdata_2016-01.bcolz') Now we have a bcolz file on disk that can be queried using [bquery](https://github.com/visualfabriq/bquery/). But we also want to show how to use the distributed functionality of bqueryd, so we split the file that we have just created into some smaller chunks. import bcolz, bquery ct = bquery.open(rootdir='tripdata_2016-01.bcolz') NR_SHARDS = 10 step = len(ct) / NR_SHARDS remainder = len(ct) % NR_SHARDS count = 0 for idx in range(0, len(ct), step): if idx == len(ct)*(NR_SHARDS-1): step = step + remainder print 'Creating file tripdata_2016-01-%s.bcolzs'%count ct_shard = bcolz.fromiter( ct.iter(idx, idx+step), ct.dtype, step, rootdir='tripdata_2016-01-%s.bcolzs'%count, mode='w' ) ct_shard.flush() count += 1 ## Running bqueryd Now to test using bqueryd. If the bqueryd was successfully installed using pip, and your virtualenvironment is activated, you should now have a script named ```bqueryd``` on your path. You can start up a controller. Before starting bqueryd, also make sure that you have a locally running [Redis](https://redis.io/) server. bqueryd controller & If you already have a controller running, you can now also run ```bqueryd``` without any arguments and it will try and connect to your controller and then drop you into an [IPython](https://ipython.org/) shell to communicate with your bqueryd cluster. bqueryd From the ipython prompt you have access to a variable named 'rpc'. (if you had at least one running controller). From the rpc variable you can execute commands to the bqueryd cluster. For example: >>> rpc.info() Will show status information on your current cluster, with only one controller node running there is not so much info yet. First exist your ipython session to the shell. Lets also start two worker nodes: bqueryd worker --data_dir=`pwd` & bqueryd worker --data_dir=`pwd` & At this point you should have a controller and two workers running in the background. When you execute ```bqueryd``` again and do: >>> rpc.info() There should be more information on the running controller plus the two worker nodes. By default worker nodes check for bcolz files in the ```/srv/bcolz/``` directory. In the previous section we ran some worker nodes with a command line argument --data_dir to use the bcolz files in the current directory. So what kind of other commands can we send to the nodes? Here are some things to try: >>> rpc.ping() >>> rpc.sleep(10) >>> rpc.loglevel('debug') >>> rpc.sleep(2) >>> rpc.loglevel('info') >>> rpc.killworkers() Notice the last command sent, this kills all the workers connected to all running controllers in the network. The controllers still keep on running. In typical setup the nodes will have been started up and kept running by a tool like [Supervisor](http://supervisord.org/) By using the 'killworkers' command it effectively re-boots all your workers. The 'sleep' call is just for testing to see if any workers are responding. The call is not performed on the caller or the connecting node, but perfomeed by a worker chosen at random. It is possible to stop all workers and all controllers in the bqueryd network by issuing the command: >>> rpc.killall() ## Configuration There is minimally **one** thing to configure to use bqueryd on a network, assuming that all other defaults are chosen. **The address of the Redis server used**. This is set in the file named ```/etc/bqueryd.cfg``` You could create this file and change the line to read, for example: redis_url = redis://127.0.0.1:6379/0 And change the IP address to the address of your running Redis instance. This needs to be done on every machine that you plan on running a bqueryd node. As a convenience there is also en example configuration file for running a bquery installation using Supervisor in [misc/supervisor.conf](misc/supervisor.conf) ## Doing calculations The whole point of having a bqueryd cluster running is doing some calculations. So once you have a controller with some worker nodes running and connected, you can drop into the bqueryd ipython shell, and for example do: >>> rpc.groupby(['tripdata_2016-01.bcolz'], ['payment_type'], ['fare_amount'], []) But we can also use the sharded data to do the same calculation: >>> import os >>> bcolzs_files = [x for x in os.listdir('.') if x.endswith('.bcolzs')] >>> rpc.groupby(bcolzs_files, ['payment_type'], [['fare_amount', 'sum', 'fare_amount']], [], aggregate=True) To see how long a rpc call took, you can check: >>> rpc.last_call_duration The sharded version actually takes longer to run than the version using the bcolz file only. But if we start up more workers, the call is speeded up. For relatively small files like in this example, the speedup is small, but for larger datasets the overhead is worthwhile. Start up a few more workers, and run the query above. ## Executing arbitrary code It is possible to your bqueryd workers to import and execute arbitrary Python code. **This is a potentially huge big security risk if you do not run your nodes on trusted servers behind a good firewall** Make sure you know what you are doing before just starting up and running bqueryd nodes. With that being said, if you have a cluster running, try something like: >>> rpc.execute_code(function='os.listdir', args=['.'], wait=True) This should pick a random worker from those connected to the controller and run the Python listdir command in the args specified. The use of this is to run code to execute other functions from the built-in bquery/bcolz aggregation logic. This enables one to perform other business specific operations over the netwok using bqueryd nodes. ## Distributing bcolz files If your system is properly configured to use [boto](https://github.com/boto/boto3) for communication with Amazon Web Services, you can use bqueryd to automically distribute collections of files to all nodes in the bqueryd cluster. Create some bcolz files in the default bqueryd directory ```/srv/bcolz/``` (or mo