在 EMR 集群上安装 pandas

Question

TLDR - 我想在 EMR 引导操作中运行命令 sudo yes | sudo pip3 uninstall numpy 两次，但它只运行一次。

我首先要说的是，我的目标是在 EMR 集群上运行支持 Pyspark 的 EMR 托管笔记本。由于各种原因，我还需要在集群上安装 pandas。首先，我遇到了一个问题，即默认 python3 安装中存在两个 numpy 版本，并且必须将它们都删除才能使用较新的版本（如在此线程中 - How do I have multiple versions of numpy installed on Amazon EMR and how to I delete the early versions?）。

如果我 ssh 进入主节点并执行两次sudo yes | sudo pip3 uninstall numpy，它可以工作：

[hadoop@ip-xxx-xx-xx-xxx ~]$ sudo yes | sudo pip3 uninstall numpy
Uninstalling numpy-1.21.1:
  /usr/bin/f2py
  /usr/local/bin/f2py
  /usr/local/bin/f2py3
  /usr/local/bin/f2py3.7
    .......
    .......
    .......
  /usr/local/lib64/python3.7/site-packages/numpy/typing/tests/test_runtime.py
  /usr/local/lib64/python3.7/site-packages/numpy/typing/tests/test_typing.py
  /usr/local/lib64/python3.7/site-packages/numpy/version.py
Proceed (y/n)?   Successfully uninstalled numpy-1.21.1
[hadoop@ip-xxx-xx-xx-xxx ~]$ sudo yes | sudo pip3 uninstall numpy
Uninstalling numpy-1.16.5:
  /usr/local/lib64/python3.7/site-packages/numpy
  /usr/local/lib64/python3.7/site-packages/numpy-1.16.5-py3.7.egg-info
Proceed (y/n)?   Successfully uninstalled numpy-1.16.5

我从 python3 安装中删除了 numpy，然后我可以正常安装 numpy 和 pandas。
当我想使用引导操作执行相同的操作时，就会出现问题。使用这个 bootstrap.sh 文件：

#!/bin/bash
sudo yes | sudo yum install python3-devel
sudo pip3 install cython
sudo pip3 install matplotlib
sudo yes | sudo pip3 uninstall numpy
sudo pip3 install pyspark boto3
sudo yes | sudo pip3 uninstall numpy
sudo pip3 install numpy
sudo pip3 install pandas

请注意，我在这里卸载了 numpy 两次，但它只是忽略了第二个 sudo yes | sudo pip3 uninstall numpy 命令！因为我没有卸载第二个 numpy 安装，所以会导致 pandas 安装损坏（再次，请参阅我之前链接到的线程）。为什么会这样？因为引导操作不起作用，并且无法通过 ssh 连接到从属节点，所以我的 pandas 安装已损坏，无法修复。

Answer 1

很遗憾，目前对于 EMR 集群无法解决此问题。 我知道它不是流行的解决方案，但您正在寻找“没有解决方案”。

您可以在此处查看官方aws论坛报告的问题以获取更多详细信息。

我将总结以下问题。

如果你想安装 pandas，你需要有 numpy.目前的问题是（基于目前的结论），无论 bootstrap.sh 中发生什么，系统 python37-numpy 包都会在 bootstrap.sh 之后自动安装

这意味着，如果您通过 bootstrap 再次安装 nympy，在 python 中导入 numpy 将导致导入旧版本，这是通过在 emr 集群设置期间 boostrap 之后发生的步骤强制执行的。

您有两种解决方案：

1. 在集群启动后手动修复问题（这意味着您需要在整个集群中手动执行此操作，或者通过 ssh 调用 bash 脚本来自动执行此过程）
2. 您根本不安装 numpy，而是使用当前版本，而是使用与该版本兼容的 pandas。

我个人使用第二种方案，安装了pandas 1.1.5，直到numpy版本被aws/emr升级。

我只想补充一下，关于您的陈述“不起作用，并且不可能通过 ssh 进入从节点”。您可以随时通过 ssh 进入主节点和执行程序。 emr 的默认安全组不公开端口 22，因此如果您无法 ssh 到执行程序，我会先看看那里。

Answer 2

好的，它不是很漂亮，但这就是我要解决的问题：
首先我使用了 EMR 版本5.33.0，并在集群上安装了以下应用程序：Spark、JupyterEnterpriseGateway（后者是因为我使用的是 EMR 笔记本）。
我使用以下引导操作为 python 安装必要的东西：

#!/bin/bash
sudo yes | sudo yum install python3-devel
sudo pip3 install cython
sudo pip3 install setuptools --upgrade

集群启动后，我运行这个外部 python 脚本，它必须提供主节点的公共 DNS 名称、主节点和从节点的私有 IP 地址以及私有密钥文件路径。它所做的是卸载 numpy，再次安装，然后安装 pandas。我不得不说这个解决方案感觉很hacky，我更喜欢通过引导操作来实现它，但我仍然不明白为什么我不能。

import argparse
import paramiko
from scp import SCPClient
import os

INSTALL_COMMANDS = [
    'sudo yes | sudo pip3 uninstall numpy',
    'sudo pip3 install numpy',
    'sudo pip3 install pandas',
]


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--master_address')
    parser.add_argument('--master_ip')
    parser.add_argument('--slave_ips', nargs='+')
    parser.add_argument('--username', default='hadoop')
    parser.add_argument('--pk_filepath')
    args = parser.parse_args()

    key = paramiko.RSAKey.from_private_key_file(args.pk_filepath)
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    client.connect(hostname=args.master_address, username=args.username, pkey=key)
    scp = SCPClient(client.get_transport())
    scp.put(args.pk_filepath, remote_path='~')
    command_chmod = f'chmod 600 {os.path.split(args.pk_filepath)[-1]}'
    print(f'executing: {command_chmod}')
    stdin, stdout, stderr = client.exec_command(command_chmod)

    all_clients = [client]
    for slave_ip in args.slave_ips:
        slave_channel = client.get_transport().open_channel('direct-tcpip',
                                                            (slave_ip, 22),
                                                            (args.master_ip, 22))
        slave_client = paramiko.SSHClient()
        slave_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        slave_client.connect(hostname=slave_ip,
                             username=args.username,
                             pkey=key,
                             sock=slave_channel)
        all_clients.append(slave_client)

    for client in all_clients:
        for command in INSTALL_COMMANDS:
            print(f'executing {command} on {client}')
            stdin, stdout, stderr = client.exec_command(command)
            print(stderr.read())