11.17【大数据】Hadoop【DEBUG】

列出hdfs文件系统所有的目录和文件

主节点上

子结点

是一样的

为什么能登进 slave 02 的主机,但是 master 当中依然显示 slave 02 为 DeadNode?

hadoop坏死节点的重启_hadoop3 子节点重启-CSDN博客

注意hadoop-daemon.sh 实际上位于 Hadoop 的 sbin 目录中，而不是 bin 目录中。因此，你需要将 sbin 目录也添加到 PATH 中

[hadoop@slave02 ~]$ echo 'export PATH=$PATH:/usr/local/hadoop/sbin' >> ~/.bashrc
[hadoop@slave02 ~]$ source ~/.bashrc

验证 hadoop-daemon.sh 路径 确认 hadoop-daemon.sh 脚本现在可以在 PATH 中找到：

[hadoop@slave02 ~]$ which hadoop-daemon.sh

bin/spark-submit --master spark://master:7077 --py-files /home/hadoop/Ex2_WordCount/WordCount.py

Error: Missing application resource 表示你在提交 Spark 作业时没有指定主应用程序文件（即你要运行的 Python 脚本）。spark-submit 命令需要知道你要运行的具体脚本文件。

bin/spark-submit --master spark://master:7077 --py-files /home/hadoop/Ex2_WordCount/WordCount.py /home/hadoop/Ex2_WordCount/WordCount.py

一直卡在正在连接远程?

pro2

No route to host-CSDN博客

sudo firewall-cmd --zone=public --query-port=36605/tcp

Caused by: io.netty.channel.AbstractChannel$AnnotatedNoRouteToHostException: No route to host: /192.168.0.132:44149
Caused by: java.net.NoRouteToHostException: No route to host

异常信息

java.io.IOException: Connection reset by peer
- 这是一个常见的网络错误，表示远程主机强制关闭了连接。具体原因可能有多种：
  - 网络问题：网络不稳定或中断，导致连接被重置。
  - 防火墙或安全软件：防火墙或安全软件阻止了连接。
  - 远程主机崩溃：远程主机可能出现了故障或崩溃，导致连接被重置。
  - 超时：连接超时，导致连接被重置。
  - 资源限制：远程主机的资源（如文件描述符）达到上限，无法处理新的连接请求。

检查防火墙和安全软件：
- 确保防火墙或安全软件没有阻止必要的端口。
- 暂时禁用防火墙或安全软件，看看问题是否解决：
  sh
  
  sudo systemctl stop firewalld

检查网络连接：
- 确保网络连接稳定，没有丢包或延迟。
- 使用 ping 或 traceroute 命令检查网络连通性：
  sh
  
  ping 124.71.188.246 traceroute 124.71.188.246

优化 Spark 配置：
- 调整 Spark 的网络超时和重试机制，以提高连接的稳定性：
  spark.network.timeout 120s
- spark.io.compression.codec lz4

调整资源限制：
- 检查远程主机的资源限制，特别是文件描述符数量：
  sh
  
  ulimit -n
- 如果需要，可以增加文件描述符的限制：
  sh
  
  ulimit -n 10000

### 分析 `ping` 结果

你提供的 `ping` 命令输出显示：

```plaintext
[hadoop@master spark]$ ping 192.168.0.249
PING 192.168.0.249 (192.168.0.249) 56(84) bytes of data.
From 192.168.0.14 icmp_seq=1 Destination Host Unreachable
From 192.168.0.14 icmp_seq=2 Destination Host Unreachable
From 192.168.0.14 icmp_seq=3 Destination Host Unreachable
From 192.168.0.14 icmp_seq=4 Destination Host Unreachable
From 192.168.0.14 icmp_seq=5 Destination Host Unreachable
From 192.168.0.14 icmp_seq=6 Destination Host Unreachable
From 192.168.0.14 icmp_seq=7 Destination Host Unreachable
From 192.168.0.14 icmp_seq=8 Destination Host Unreachable
```

### 解释

- **Destination Host Unreachable**：这表示目标主机不可达。通常这意味着目标主机（192.168.0.249）没有响应，或者中间的路由器无法将数据包转发到目标主机。
- **From 192.168.0.14**：这表示 ICMP 回复来自中间的路由器（192.168.0.14），而不是目标主机。这表明问题可能出在目标主机本身，或者目标主机与路由器之间的连接有问题。

### 可能的原因及解决方法

1. **目标主机未启动或网络未连接**：
   - **检查目标主机**：确保目标主机（192.168.0.249）已经启动，并且网络连接正常。
   - **登录目标主机**：
     ```sh
     ssh hadoop@192.168.0.249
     ```

2. **目标主机的网络配置问题**：
   - **检查网络配置**：确保目标主机的网络配置正确，特别是 IP 地址和网关设置。
   - **查看网络接口状态**：
     ```sh
     ip addr show
     ```

3. **中间路由器问题**：
   - **检查中间路由器**：确保中间路由器（192.168.0.14）工作正常，没有配置错误。
   - **重启路由器**：
     ```sh
     sudo systemctl restart networking
     ```

4. **防火墙或安全组设置**：
   - **检查防火墙设置**：确保防火墙没有阻止 ICMP 请求。
   - **临时禁用防火墙**：
     ```sh
     sudo systemctl stop firewalld
     ```

### 示例

1. **检查目标主机状态**：
   ```sh
   ssh hadoop@192.168.0.249
   ```

2. **查看目标主机的网络配置**：
   ```sh
   ip addr show
   ```

3. **检查中间路由器状态**：
   ```sh
   ssh hadoop@192.168.0.14
   sudo systemctl status networking
   ```

4. **临时禁用防火墙**：
   ```sh
   sudo systemctl stop firewalld
   ```

### 结论

`Destination Host Unreachable` 错误通常表示目标主机不可达，可能是由于目标主机未启动、网络配置错误、中间路由器问题或防火墙设置不当。通过上述步骤，你可以逐步排查并解决这个问题。如果问题依然存在，建议进一步检查集群的网络配置和日志，以获取更多线索。

### 日志解释

你提供的日志信息如下：

```plaintext
24/11/17 16:07:28 WARN TaskSetManager: Lost task 19.0 in stage 3.1 (TID 103) (192.168.0.14 executor 1): FetchFailed(BlockManagerId(3, 192.168.0.249, 42553, None), shuffleId=0, mapIndex=1, mapId=9, reduceId=19, message=
org.apache.spark.shuffle.FetchFailedException
        at org.apache.spark.errors.SparkCoreErrors$.fetchFailedError(SparkCoreErrors.scala:437)
        at org.apache.spark.storage.ShuffleBlockFetcherIterator.throwFetchFailedException(ShuffleBlockFetcherIterator.scala:1239)
        at org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:971)
        at org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:86)
        at org.apache.spark.util.CompletionIterator.next(CompletionIterator.scala:29)
        at scala.collection.Iterator$$anon$11.nextCur(Iterator.scala:486)
        at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:492)
        at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)
        at org.apache.spark.util.CompletionIterator.hasNext(CompletionIterator.scala:31)
        at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
        at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)
        at scala.collection.Iterator.foreach(Iterator.scala:943)
        at scala.collection.Iterator.foreach$(Iterator.scala:943)
        at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
        at org.apache.spark.api.python.PythonRDD$.writeIteratorToStream(PythonRDD.scala:322)
        at org.apache.spark.api.python.PythonRunner$$anon$2.writeIteratorToStream(PythonRunner.scala:751)
        at org.apache.spark.api.python.BasePythonRunner$WriterThread.$anonfun$run$1(PythonRunner.scala:451)
        at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:1928)
        at org.apache.spark.api.python.BasePythonRunner$WriterThread.run(PythonRunner.scala:282)
Caused by: java.io.IOException: Connecting to /192.168.0.249:42553 failed in the last 4750 ms, fail this connection directly
        at org.apache.spark.network.client.TransportClientFactory.createClient(TransportClientFactory.java:210)
        at org.apache.spark.network.netty.NettyBlockTransferService$$anon$2.createAndStart(NettyBlockTransferService.scala:131)
        at org.apache.spark.network.shuffle.RetryingBlockTransferor.transferAllOutstanding(RetryingBlockTransferor.java:173)
        at org.apache.spark.network.shuffle.RetryingBlockTransferor.lambda$initiateRetry$0(RetryingBlockTransferor.java:206)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
        at java.lang.Thread.run(Thread.java:750)
```

### 解释

1. **警告信息**：
   ```plaintext
   24/11/17 16:07:28 WARN TaskSetManager: Lost task 19.0 in stage 3.1 (TID 103) (192.168.0.14 executor 1)
   ```
   - 这表示在阶段 3.1 中，任务 19.0（任务 ID 103）在执行器 1 上丢失。

2. **异常信息**：
   ```plaintext
   FetchFailed(BlockManagerId(3, 192.168.0.249, 42553, None), shuffleId=0, mapIndex=1, mapId=9, reduceId=19, message=
   org.apache.spark.shuffle.FetchFailedException
   Caused by: java.io.IOException: Connecting to /192.168.0.249:42553 failed in the last 4750 ms, fail this connection directly
   ```
   - `FetchFailedException` 表示在从 BlockManagerId 为 `(3, 192.168.0.249, 42553, None)` 的节点获取 shuffle 数据时失败。
   - 具体的 shuffle ID 为 0，mapIndex 为 1，mapId 为 9，reduceId 为 19。
   - `java.io.IOException` 表示在连接到 `192.168.0.249:42553` 时失败，连接尝试在 4750 毫秒内未成功。

### 常见原因及解决方法

1. **网络问题**：
   - **检查网络连接**：确保所有节点之间的网络连接稳定，没有丢包或延迟。
   - **使用 `ping` 和 `traceroute` 命令**：
     ```sh
     ping 192.168.0.249
     traceroute 192.168.0.249
     ```

2. **防火墙或安全软件**：
   - **检查防火墙设置**：确保防火墙或安全软件没有阻止必要的端口。
   - **暂时禁用防火墙**：
     ```sh
     sudo systemctl stop firewalld
     ```

3. **资源不足**：
   - **检查资源使用情况**：确保执行器有足够的资源（如内存、CPU）来处理任务。
   - **调整 Spark 配置**：
     ```properties
     spark.executor.memory 4g
     spark.executor.cores 2
     spark.network.timeout 120s
     spark.shuffle.io.maxRetries 10
     spark.shuffle.io.retryWait 10s
     ```

4. **BlockManager 故障**：
   - **检查 BlockManager 日志**：查看 BlockManager 的日志，寻找异常信息。
   - **重启 BlockManager**：如果 BlockManager 出现故障，可以尝试重启它。

5. **Shuffle 文件损坏**：
   - **删除 Shuffle 文件**：删除有问题的 Shuffle 文件，让 Spark 重新生成。
   - **调整 Shuffle 文件存储位置**：
     ```properties
     spark.local.dir /tmp/spark-local-dir
     ```

### 示例

假设你已经检查了网络连接和资源使用情况，接下来可以尝试以下步骤：

1. **检查网络连接**：
   ```sh
   ping 192.168.0.249
   traceroute 192.168.0.249
   ```

2. **调整 Spark 配置**：
   编辑 `conf/spark-defaults.conf` 文件，增加以下配置：
   ```properties
   spark.executor.memory 4g
   spark.executor.cores 2
   spark.network.timeout 120s
   spark.shuffle.io.maxRetries 10
   spark.shuffle.io.retryWait 10s
   ```

3. **重新提交 Spark 作业**：
   ```sh
   bin/spark-submit --master spark://master:7077 --py-files /home/hadoop/Ex2_WordCount/WordCount.py /home/hadoop/Ex2_WordCount/WordCount.py
   ```

### 结论

`FetchFailedException` 通常与网络问题、防火墙设置、资源不足、BlockManager 故障或 Shuffle 文件损坏有关。通过上述步骤，你可以逐步排查和解决这个问题。如果问题依然存在，建议进一步检查集群的配置和日志，以获取更多线索。

nmap -p 7077 master

为什么在WEBUI当中显示有worker而且Alive，但是在日志中一直显示报错？

为什么在 WEBUI 当中有 alive 的 worker,是私网 ip,但是在终端缺 ping 不通?

./sbin/start-all.sh -h master

./sbin/start-all.sh spark://master_ip:7077

./sbin/sto-all.sh -h master

Caused by: io.netty.channel.AbstractChannel$AnnotatedNoRouteToHostException: No route to host: /192.168.0.132:36645
Caused by: java.net.NoRouteToHostException: No route to host
        at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
        at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:716)
        at io.netty.channel.socket.nio.NioSocketChannel.doFinishConnect(NioSocketChannel.java:337)
        at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.finishConnect(AbstractNioChannel.java:334)
        at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:776)
        at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:724)
        at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:650)
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562)
        at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
        at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
        at java.lang.Thread.run(Thread.java:750)

根据您提供的代码和错误信息，问题仍然出在加载 `bert-base-chinese` 模型的 `BertTokenizer` 时失败。为了确保在分布式环境下能够成功加载模型，我们可以采取以下措施：

1. **确保所有节点上都安装了 `transformers` 和 `torch` 库**。
2. **预先下载模型文件并分发到所有节点**。
3. **使用 `--files` 参数将模型文件与 Spark 任务一起分发**。

### 1. 安装必要的库

确保所有节点上都安装了 `transformers` 和 `torch` 库：

```sh
pip install transformers torch
```

### 2. 预先下载模型文件

在一台机器上下载 `bert-base-chinese` 模型文件，并将其分发到所有节点。可以使用以下命令下载模型：

```sh
mkdir -p /path/to/models/bert-base-chinese
transformers-cli download bert-base-chinese --cache-dir /path/to/models/bert-base-chinese
```

### 3. 修改代码以加载本地模型文件

修改代码，使其从本地路径加载模型文件：

```python
from pyspark import SparkConf, SparkContext
from transformers import BertTokenizer, BertModel
import torch
import jieba
import re

DATAPATH = 'hdfs://master:9000/ex2/stop_words.txt'
SAVAPATH = '/home/hadoop/Ex2_WordCount/results/res.txt'
# 加载 BERT 模型和分词器
model_name = '/path/to/models/bert-base-chinese'
tokenizer = BertTokenizer.from_pretrained(model_name)
model = BertModel.from_pretrained(model_name)

# 打印模型结构
print(model)

# 打印模型参数
for name, param in model.named_parameters():
    print(f"Parameter Name: {name}, Shape: {param.shape}")

# 打印模型配置
print(model.config)

conf = SparkConf().setAppName("ex2").setMaster("spark://master:7077").set("spark.executor.memory","4g")
sc = SparkContext(conf=conf)

def getStopWords(stopWords_filePath):
    stopwords = sc.textFile(stopWords_filePath).collect()
    return stopwords

def bertTokenize(text):
    # 使用 BERT 分词器进行分词
    tokens = tokenizer.tokenize(text)
    return tokens

def jiebaCut(answers_filePath):
    # 读取 answers.txt
    answersRdd = sc.textFile(answers_filePath)  # answersRdd 每一个元素对应 answers.txt 每一行
    # 利用 SparkRDD reduce() 函数, 合并所有回答
    str = answersRdd.reduce(lambda a, b: a + b)
    # 使用 BERT 分词器进行分词
    words_list = bertTokenize(str)
    return words_list

def wordcount():
    # 读取停用词表
    stopwords = getStopWords(DATAPATH)
    # 使用 BERT 分词
    words_list = jiebaCut(SAVAPATH)

    # 词频统计
    wordsRdd = sc.parallelize(words_list)
    
    resRdd = wordsRdd.filter(lambda word: word not in stopwords) \
                     .filter(lambda word: len(word) > 1) \
                     .map(lambda word: (word, 1)) \
                     .reduceByKey(lambda a, b: a + b) \
                     .sortBy(lambda x: x[1], ascending=False)
    return resRdd

if __name__ == '__main__':
    # 进行词频统计并可视化
    resRdd = wordcount()
    print(resRdd.take(10))  # 查看前10个
```

### 4. 提交 Spark 任务

使用 `spark-submit` 提交任务，并确保将模型文件与任务一起分发：

```sh
spark-submit --master spark://master:7077 --files /path/to/models/bert-base-chinese/vocab.txt /home/hadoop/Ex2_WordCount/WordCountByBert.py
```

### 注意事项

1. **模型文件路径**：
   - 确保 `model_name` 路径正确指向本地模型文件的目录。
   - 如果模型文件较大，建议使用 `rsync` 或其他工具将模型文件分发到所有节点。

2. **资源管理**：
   - 确保 Spark 配置有足够的资源（内存和 CPU）来处理 BERT 模型的计算需求。

3. **性能优化**：
   - 在分布式环境中，频繁加载模型可能会导致性能下降。可以考虑将模型加载逻辑移到 `mapPartitions` 中，每个分区只加载一次模型。

通过以上步骤，您应该能够在 Hadoop + Spark 环境中成功使用 BERT 模型进行分词。希望这些信息对您有所帮助！