有没有办法限制默认线程调度程序使用的内核数(使用dask数据帧时默认)？

使用compute,您可以使用以下命令指定它:

df.compute(get=dask.threaded.get, num_workers=20)

但我想知道是否有办法将其设置为默认值,因此您不需要为每次compute调用指定此项？

例如,在小型集群(例如64个核心)的情况下会很有趣,但是与其他人共享(没有作业系统),并且我不希望在使用dask开始计算时占用所有核心.

我有一个简单的功能来排名扑克手(手是字符串).

我打电话给它,rA,rB = rank(a),rank(b)这是我的实现.在没有@jit(nopython = True)的情况下运行良好,但有了它,它会失败:

   File "C:/Users/avi_na/Desktop/poker.py", line 190, in <module>
        rA,rB = rank(a),rank(b)

      File "C:\Continuum\anaconda3\lib\site-packages\numba\dispatcher.py", line 344, in _compile_for_args
        reraise(type(e), e, None)

      File "C:\Continuum\anaconda3\lib\site-packages\numba\six.py", line 658, in reraise
        raise value.with_traceback(tb)

TypingError: cannot determine Numba type of <class 'builtin_function_or_method'>

from numba import jit
from numba.types import string

@jit(nopython=True)
def rank(hand):
#    assert(len(hand) == 5)
    rank = "N/A"

    p = pd.Series([h[0] for h in hand]).value_counts()
    v = sorted(set(pd.Series([h[0] for h in hand]).values), reverse=True)
    s = sorted(hand, key=lambda k:k[0]) …

介绍

目前，我正在尝试将 dask 与 gensim 配合使用来进行 NLP 文档计算，并且在将我的语料库转换为“ TaggedDocument ”时遇到问题。

因为我尝试了很多不同的方法来解决这个问题，所以我将列出我的尝试。

每次处理这个问题的尝试都会遇到略有不同的困境。

首先是一些初步的假设。

数据

df.info()
<class 'dask.dataframe.core.DataFrame'>
Columns: 5 entries, claim_no to litigation
dtypes: object(2), int64(3)

  claim_no   claim_txt I                                    CL ICC lit
0 8697278-17 battery comprising interior battery active ele... 106 2 0

所需输出

>>tagged_document[0]
>>TaggedDocument(words=['battery', 'comprising', 'interior', 'battery', 'active', 'elements', 'battery', 'cell', 'casing', 'said', 'cell', 'casing', 'comprising', 'first', 'casing', 'element', 'first', 'contact', 'surface', 'second', 'casing', 'element', 'second', 'contact', 'surface', 'wherein', 'assembled', 'position', 'first', 'second', 'contact', 'surfaces', 'contact', 'first', 'second', …

因此，我想并行地遍历pandas df，所以假设我有15行，那么我想并行地遍历它，而不是一个接一个地迭代。

df：-

df = pd.DataFrame.from_records([
    {'domain':'dnd','duration':'90','media_file':'testfont.wav','user':'tester_food','channel':'confctl-2' },
    {'domain':'hrpd','duration':'90','media_file':'testfont.wav','user':'tester_food','channel':'confctl-2' },
    {'domain':'blhp','duration':'90','media_file':'testfont.wav','user':'tester_food','channel':'confctl-2' },
    {'domain':'rbswp','duration':'90','media_file':'testfont.wav','user':'tester_food','channel':'confctl-2' },
    {'domain':'foxbp','duration':'90','media_file':'testfont.wav','user':'tester_food','channel':'confctl-2' },
    {'domain':'rbsxbp','duration':'90','media_file':'testfont.wav','user':'tester_food','channel':'confctl-2' },
    {'domain':'dnd','duration':'90','media_file':'testfont.wav','user':'tester_food','channel':'confctl-2' },
    {'domain':'hrpd','duration':'90','media_file':'testfont.wav','user':'tester_food','channel':'confctl-2' }

])

所以，我遍历df并制作命令行，然后将输出存储在df中并进行数据过滤，最后将其存储到influxdb中。问题是我在迭代过程中一个接一个地做。我想并行遍历所有行。

截至目前，我已经制作了20个脚本，并使用多处理并行处理所有脚本。当我必须在所有20个脚本中进行更改时，这是一种痛苦。我的脚本如下所示：-

for index, row in dff.iterrows():
    domain = row['domain']
    duration = str(row['duration'])
    media_file = row['media_file']
    user = row['user']
    channel = row['channel']
    cmda = './vaa -s https://' + domain + '.www.vivox.com/api2/ -d ' + 
    duration + ' -f ' + media_file + ' -u .' + user + '. -c 
    sip:confctl-2@' + domain + '.localhost.com -ati 0ps-host …