मैंने हाल ही में एक अध्ययन किया जिसमें इनपुट डेटा के कई सौ हजार सेटों को संसाधित करना आवश्यक था। प्रत्येक सेट के लिए - कुछ गणना करें, सभी गणनाओं के परिणामों को एक साथ इकट्ठा करें और कुछ मानदंडों के अनुसार "सर्वश्रेष्ठ" का चयन करें। संक्षेप में, यह ब्रूटफोर्स बस्ट है। एक ही बात का उपयोग करते हुए एमएल मॉडल के मापदंडों का चयन करते समय होता है GridSearch।

हालाँकि, कुछ बिंदु से, गणनाओं का आकार एक कंप्यूटर के लिए बहुत बड़ा हो सकता है, भले ही आप इसका उपयोग करके कई प्रक्रियाओं में चलाते हों joblib। या, अधिक सटीक रूप से, यह एक अधीर प्रयोगकर्ता के लिए बहुत लंबा हो जाता है।

और चूंकि एक आधुनिक अपार्टमेंट में आप अब एक से अधिक "अंडरलोड" कंप्यूटर पा सकते हैं, और कार्य स्पष्ट रूप से बड़े पैमाने पर संगामिति के लिए उपयुक्त है - यह आपके होम क्लस्टर को इकट्ठा करने और उस पर इस तरह के कार्यों को चलाने का समय है।

Dask लाइब्रेरी ( https://dask.org/ ) "होम क्लस्टर" बनाने के लिए एकदम सही है । नोड्स पर मांग करना और स्थापित करना आसान है, जो क्लस्टर कंप्यूटिंग में "प्रवेश स्तर" को गंभीरता से कम करता है।

अपने क्लस्टर को कॉन्फ़िगर करने के लिए, आपको सभी कंप्यूटरों की आवश्यकता है:

• अजगर दुभाषिया स्थापित करें
• dask
• (scheduler) (worker)

, , — , .

(https://docs.dask.org/) . , .

python

Dask pickle, , python.
3.6 3.7 , . 3.8 - pickle.

" ", , , .

Dask

Dask pip conda

pip install dask distributed bokeh

dask, bokeh , , "-" dask dashboard.
. .

gcc, :

• MacOS xcode
• docker image docker-worker, "" , python:3.6-slim-buster . , python:3.6.

dask

- . . — .

$ dask-scheduler

- , .

$ dask-worker schedulerhost:8786 --nprocs 4 --nthreads 1 --memory-limit 1GB --death-timeout 120 -name MyWorker --local-directory /tmp/

• nprocs / nthreads — , , . GIL -, - , numpy. .
• memory-limit — , . — - , . , .
• death-timeout — , - , . -. , , .
• name — -, . , "" -.
• local-directory — ,

- Windows

, dask-worker . , , dask-worker .

" " . NSSM (https://www.nssm.cc/).

NSSM, , , . , , - . NSSM .

NSSM . " "

Firewall

firewall: -.

, , -, — . , — . , , .

- . , .

Dask

:

from dask.distributed import Client

client = Client('scheduler_host:port')

— "" , .

, , . pandas, numpy, scikit-learn, tensorflow.
, .

, ? — pip

def install_packages():
    try:
        import sys, subprocess
        subprocess.check_call([sys.executable, '-m', 'pip', 'install', 'mypackage'])
        return (0)
    except:
        return (1)

from dask.distributed import Client

client = Client('scheduler:8786')
client.run(install_packages)

, , . . "" , , .

, , — .
, , Dask .

. Client upload_file(). , .

- , zip.

from dask.distributed import Client 
import numpy as np
from my_module import foo
from my_package import bar 

def zoo(x)
  return (x**2 + 2*x + 1)

x = np.random.rand(1000000)

client = Client('scheduler:8786')

#        . 
#     
r3 = client.map(zoo, x) 

#  foo  bar     ,
#            
client.upload_file('my_module.py')
client.upload_file('my_package.zip')

#    
r1 = client.map(foo, x)
r2 = client.map(bar, x) 

joblib

joblib . joblib — :

joblib

from joblib import Parallel, delayed
...
res = Parallel(n_jobs=-1)(delayed(my_proc)(c, ref_data) for c in candidates)

joblib + dask

# 
from joblib import Parallel, delayed, parallel_backend
from dask.distributed import Client
...
client = Client('scheduler:8786')

with parallel_backend('dask'): #  ""    
  res = Parallel(n_jobs=-1)(delayed(my_proc)(c, ref_data) for c in candidates)

, , . , — :

16 , .

— 10-20 , 200.

, - .

# 
from joblib import Parallel, delayed, parallel_backend
from dask.distributed import Client
...
client = Client('scheduler:8786')

with parallel_backend('dask', scatter = [ref_data]):
  res = Parallel(n_jobs=-1, batch_size=<N>, pre_dispatch='3*n_jobs')(delayed(my_proc)(c, ref_data) for c in candidates)

batch_size. — , , .
pre_dispatch.

, .

• — . , .
• — ( )
• —

3.5-4 , . : , - , , .

, batch_size pre_dispatch . 8-10 .

, , - (, , ), scatter. .

, .

GridSearchCV

scikit-learn joblib , — dask

:

...

lr = LogisticRegression(C=1, solver="liblinear", penalty='l1', max_iter=300)

grid = {"C": 10.0 ** np.arange(-2, 3)}

cv = GridSearchCV(lr, param_grid=grid, n_jobs=-1, cv=3, 
                  scoring='f1_weighted', 
                  verbose=True, return_train_score=True )

client = Client('scheduler:8786')

with joblib.parallel_backend('dask'):
    cv.fit(x1, y)

clf = cv.best_estimator_
print("Best params:", cv.best_params_)
print("Best score:", cv.best_score_)

:

Fitting 3 folds for each of 5 candidates, totalling 15 fits
[Parallel(n_jobs=-1)]: Using backend DaskDistributedBackend with 12 concurrent workers.
[Parallel(n_jobs=-1)]: Done   8 out of  15 | elapsed:  2.0min remaining:  1.7min
[Parallel(n_jobs=-1)]: Done  15 out of  15 | elapsed: 16.1min finished
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/sklearn/linear_model/_logistic.py:1539: UserWarning: 'n_jobs' > 1 does not have any effect when 'solver' is set to 'liblinear'. Got 'n_jobs' = 16.
  " = {}.".format(effective_n_jobs(self.n_jobs)))
Best params: {'C': 10.0}
Best score: 0.9748830491726451

dask. -.
— .

, ( ) — . — .

टास्क लाइब्रेरी विशिष्ट कार्यों के लिए एक बेहतरीन स्केलिंग टूल है। यहां तक ​​कि अगर आप केवल मूल dask.distributed का उपयोग करते हैं और विशेष एक्सटेंशन dask.dataframe, dask.array, dask.ml को छोड़ देते हैं - तो आप प्रयोगों को तेज कर सकते हैं। कुछ मामलों में, गणना के लगभग रैखिक त्वरण को प्राप्त किया जा सकता है।

और यह सब इस बात पर आधारित है कि आपके पास पहले से ही घर पर क्या है, और वीडियो देखने, अंतहीन समाचार फीड या गेम के माध्यम से स्क्रॉल करने के लिए उपयोग किया जाता है। इन संसाधनों का पूरा उपयोग करें!