我想提起这件事,只是因为它很疯狂.也许韦斯有一些想法.该文件非常规则:1100行x~3M列,数据以制表符分隔,仅由整数0,1和2组成.显然,这不是预期的.

如果我预先填充下面的数据帧,它会占用大约26GB的RAM.

h = open("ms.txt")
header = h.readline().split("\t")
h.close()
rows=1100
df = pd.DataFrame(columns=header, index=range(rows), dtype=int)

系统信息:

• python 2.7.9
• ipython 2.3.1
• numpy 1.9.1
• 大熊猫0.15.2.

欢迎任何想法.

背景

我正在研究使用芹菜(3.1.8)来处理每个文本文件(~30GB).这些文件采用fastq格式,包含大约118M的测序"读数",基本上每个都是标题,DNA序列和质量字符串的组合.此外,这些序列来自配对末端测序运行,因此我同时迭代两个文件(通过itertools.izip).我希望能够做的是将每对读取,发送到队列,并在我们集群中的一台机器上处理它们(不关心哪些)以返回清理后的版本如果清洁需要发生(例如,基于质量).

我已经建立了芹菜和兔子,我的工作人员如下:

celery worker -A tasks --autoreload -Q transient 

并配置如下:

from kombu import Queue

BROKER_URL = 'amqp://guest@godel97'
CELERY_RESULT_BACKEND = 'rpc'
CELERY_TASK_SERIALIZER = 'pickle'
CELERY_RESULT_SERIALIZER = 'pickle'
CELERY_ACCEPT_CONTENT=['pickle', 'json']
CELERY_TIMEZONE = 'America/New York'
CELERY_ENABLE_UTC = True
CELERYD_PREFETCH_MULTIPLIER = 500

CELERY_QUEUES = (
    Queue('celery', routing_key='celery'),
    Queue('transient', routing_key='transient',delivery_mode=1),
)

我选择使用rpc后端和pickle序列化来提高性能,以及不在"瞬态"队列中写入任何内容(通过delivery_mode).

芹菜创业

要设置芹菜框架,我首先在64路盒子上启动rabbitmq服务器(3.2.3,Erlang R16B03-1),将日志文件写入fast/tmp磁盘.工作进程(如上所述)在群集上的每个节点(大约34个)上启动,范围从8路到64路SMP,总共688个核心.因此,我有大量可用的CPU供工作人员用于处理队列.

工作提交/表现

芹菜启动并运行后,我通过ipython笔记本提交作业,如下所示:

files = [foo, bar]
f1 = open(files[0])
f2 = open(files[1])
res = []
count = 0
for r1, r2 in izip(FastqGeneralIterator(f1), FastqGeneralIterator(f2)):
    count += 1 …

我正在尝试模拟计算环境的创建，这需要一些其他资源，即 IAM 实例配置文件和服务角色。但是，当我创建这些 IAM 资源，然后尝试在创建计算环境中使用它们时，事情会失败，并显示：

<Message>Role arn:aws:iam::123456789012:instance-profile/InstanceProfile not found</Message>

代码如下：

@mock_batch
@mock_iam
def test_create_compute_environment(lims_objs):
    client = boto3.client("batch")
    iam = boto3.resource("iam")
    service_role = iam.create_role(
        RoleName="BatchServiceRole", AssumeRolePolicyDocument="AWSBatchServiceRole"
    )
    instance_profile = iam.create_instance_profile(
        InstanceProfileName="InstanceProfile"
    )
    instance_profile.add_role(RoleName=service_role.name)
    for elem in iam.instance_profiles.all():
        print(elem, elem.arn)

    for elem in iam.roles.all():
        print(elem)

    response = client.create_compute_environment(
        computeEnvironmentName="compute_environment",
        type="MANAGED",
        state="ENABLED",
        computeResources={
            "type": "EC2",
            "minvCpus": 0,
            "maxvCpus": 256,
            "desiredvCpus": 2,
            "instanceTypes": ["optimal"],
            "imageId": "test",
            "subnets": [],
            "securityGroupIds": [],
            "ec2KeyPair": "",
            "instanceRole": instance_profile.arn,
            "tags": {},
        },
        serviceRole=service_role.arn,
    )

在测试中，我可以看到 IAM 对象的打印，因此我知道它们正在被创建。这些是不是在摩托车模拟中共享？

iam.InstanceProfile(name='InstanceProfile') arn:aws:iam::123456789012:instance-profile/InstanceProfile …