From 5fb2d08119e53256b051fdd92dd555ab9006eb81 Mon Sep 17 00:00:00 2001 From: chupei Date: Tue, 9 Dec 2025 20:49:31 +0800 Subject: [PATCH 1/4] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D$$$=20=E7=AD=89?= =?UTF-8?q?=E4=B8=8D=E5=AE=8C=E6=95=B4=E6=95=B0=E5=AD=A6=E6=A0=87=E8=AE=B0?= =?UTF-8?q?=E8=A2=AB=E9=94=99=E8=AF=AF=E8=AF=86=E5=88=AB=E4=B8=BA=E5=85=AC?= =?UTF-8?q?=E5=BC=8F=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .pre-commit-config.yaml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index c0c19fe1..d043a140 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -51,8 +51,8 @@ repos: - mdformat_frontmatter - linkify-it-py exclude: '^tests/.*/assets/|llm_web_kit/model/assets/.*' - - repo: https://github.com/myint/docformatter - rev: v1.3.1 + - repo: https://github.com/PyCQA/docformatter + rev: v1.7.5 hooks: - id: docformatter args: [ "--in-place", "--wrap-descriptions", "119" ] From c3b9a1aeb48f585a6efe17e4970fcc045834032f Mon Sep 17 00:00:00 2001 From: chupei Date: Tue, 9 Dec 2025 20:58:45 +0800 Subject: [PATCH 2/4] x --- .pre-commit-config.yaml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index d043a140..c0c19fe1 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -51,8 +51,8 @@ repos: - mdformat_frontmatter - linkify-it-py exclude: '^tests/.*/assets/|llm_web_kit/model/assets/.*' - - repo: https://github.com/PyCQA/docformatter - rev: v1.7.5 + - repo: https://github.com/myint/docformatter + rev: v1.3.1 hooks: - id: docformatter args: [ "--in-place", "--wrap-descriptions", "119" ] From 266dd8f234be6cb8a85255873b716fad983a3fb3 Mon Sep 17 00:00:00 2001 From: chupei Date: Fri, 26 Dec 2025 15:27:36 +0800 Subject: [PATCH 3/4] fix: error when width of math img is float type --- .pre-commit-config.yaml | 10 ++-- .../html/recognizer/cc_math/tag_img.py | 6 ++- tests/llm_web_kit/simple/test_simple.py | 51 +++++++++++++++++++ 3 files changed, 61 insertions(+), 6 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index c0c19fe1..e4eab679 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -51,11 +51,11 @@ repos: - mdformat_frontmatter - linkify-it-py exclude: '^tests/.*/assets/|llm_web_kit/model/assets/.*' - - repo: https://github.com/myint/docformatter - rev: v1.3.1 - hooks: - - id: docformatter - args: [ "--in-place", "--wrap-descriptions", "119" ] + # - repo: https://github.com/myint/docformatter + # rev: v1.3.1 + # hooks: + # - id: docformatter + # args: [ "--in-place", "--wrap-descriptions", "119" ] - repo: local hooks: - id: clear-jupyter-notebook-output diff --git a/llm_web_kit/extractor/html/recognizer/cc_math/tag_img.py b/llm_web_kit/extractor/html/recognizer/cc_math/tag_img.py index e1f500c9..ae9a8700 100644 --- a/llm_web_kit/extractor/html/recognizer/cc_math/tag_img.py +++ b/llm_web_kit/extractor/html/recognizer/cc_math/tag_img.py @@ -1,3 +1,4 @@ +import re from urllib.parse import unquote from lxml.html import HtmlElement @@ -45,7 +46,10 @@ def is_display_mode(node, src_name): return True # 4. 检查图片尺寸 - if node.get('width') and int(node.get('width', '0')) > 100: + width_str = node.get('width', '') + # 提取数字部分，处理带单位的情况（如 "100px"） + width_match = re.match(r'^(\d+)', width_str) + if width_match and int(width_match.group(1)) > 100: return True # 5. 检查是否后面紧跟
标签 diff --git a/tests/llm_web_kit/simple/test_simple.py b/tests/llm_web_kit/simple/test_simple.py index 442af4a2..d5305894 100644 --- a/tests/llm_web_kit/simple/test_simple.py +++ b/tests/llm_web_kit/simple/test_simple.py @@ -718,6 +718,56 @@ def test_extract_main_html_with_table_with_math(self): self.assertIn('| $n$ | $785$ | $885$ | $1667$ |', md) self.assertIn('| $\\chi(n)$ | $e\\left(\\frac{3}{4}\\right)$ | $e\\left(\\frac{2}{3}\\right)$ | $-1$ |', md) + def test_extract_main_html_with_math_img_width_various_formats(self): + """测试img标签width属性各种格式的情况，验证不会抛出异常.""" + main_html = r''' +

Some text with inline formula:

+ + $E=mc^2$ +

And a larger image:

+ + large image +

Image with percent width:

+ + $a^2+b^2=c^2$ +

Image with float width:

+ + $x^n$ +

Image with float width and unit:

+ + $y^m$ +

Image with auto width:

+ + $z^k$ +

Image with em unit:

+ + $w^j$ +

Image with empty width:

+ + $v^i$ + ''' + + # 这个测试主要验证不会因为各种 width 值而抛出异常 + md = extract_content_from_main_html(self.url, main_html) + print(md) + + # 验证文本内容存在 + self.assertIn('Some text with inline formula', md) + self.assertIn('Image with float width', md) + self.assertIn('Image with auto width', md) + + # 验证 img 中的数学公式被正确提取 + # width <= 100 的是行内公式 $...$ + self.assertIn('$E=mc^2$', md) # width="50px", 50 <= 100 + self.assertIn('$a^2+b^2=c^2$', md) # width="80%", 80 <= 100 + self.assertIn('$z^k$', md) # width="auto", 无数字 + self.assertIn('$w^j$', md) # width="10em", 10 <= 100 + self.assertIn('$v^i$', md) # width="", 空值 + + # width > 100 的是行间公式 $$...$$ (多行格式) + self.assertIn('$$\nx^n\n$$', md) # width="512.123", 512 > 100 + self.assertIn('$$\ny^m\n$$', md) # width="123.456px", 123 > 100 + def test_extract_magic_html_with_mathjax(self): """测试包含MathJax数学公式的HTML内容提取.""" raw_html = r''' @@ -752,3 +802,4 @@ def test_extract_magic_html_with_mathjax(self): if __name__ == '__main__': unittest.main(verbosity=2) + TestSimple().test_extract_main_html_with_math_img_width_various_formats() From 84b0fbafa683733d06b77f7109344323715a8e92 Mon Sep 17 00:00:00 2001 From: chupei Date: Wed, 19 Aug 2026 17:01:38 +0800 Subject: [PATCH 4/4] feat: add pjcc prededup --- .gitignore | 1 - jupyter/pjcc-prededup/cc_dedup_fir.ipynb | 274 +++++++++++++++ jupyter/pjcc-prededup/cc_dedup_sec.ipynb | 421 +++++++++++++++++++++++ jupyter/pjcc-prededup/cc_dedup_thr.ipynb | 348 +++++++++++++++++++ 4 files changed, 1043 insertions(+), 1 deletion(-) create mode 100644 jupyter/pjcc-prededup/cc_dedup_fir.ipynb create mode 100644 jupyter/pjcc-prededup/cc_dedup_sec.ipynb create mode 100644 jupyter/pjcc-prededup/cc_dedup_thr.ipynb diff --git a/.gitignore b/.gitignore index 7457a4ea..20e6b3d1 100644 --- a/.gitignore +++ b/.gitignore @@ -11,7 +11,6 @@ __pycache__ *.pyc .vscode debug/ -*.ipynb .idea .python-version diff --git a/jupyter/pjcc-prededup/cc_dedup_fir.ipynb b/jupyter/pjcc-prededup/cc_dedup_fir.ipynb new file mode 100644 index 00000000..97fc8a13 --- /dev/null +++ b/jupyter/pjcc-prededup/cc_dedup_fir.ipynb @@ -0,0 +1,274 @@ +{ + "cells": [ + { + "cell_type": "code", + "execution_count": null, + "id": "0", + "metadata": {}, + "outputs": [], + "source": [ + "# # 获取 cc warc path list\n", + "# warc_paths = []\n", + "# for dump in DUMPS:\n", + "# dump_path = f'{CC_WARC}{dump}/'\n", + "# warc_paths.extend([x for x in list(list_s3_objects(dump_path, recursive=True)) if x.endswith('.jsonl.gz')])\n", + "\n", + "index_path = \"s3://qa-huawei/chupei/cc-domain-centric-store/PJCC/FILELIST/pjcc-diff-0628.txt\"\n", + "index_df = read_any_path(spark, index_path, config)\n", + "warc_paths = index_df.rdd.map(lambda row: row.value).collect()\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "1", + "metadata": {}, + "outputs": [], + "source": [ + "from pyspark.sql import Row\n", + "from xinghe.spark import *\n", + "from app.common.json_util import *\n", + "from xinghe.s3 import *\n", + "from pyspark.sql.types import StructType, StructField, StringType\n", + "import re\n", + "import hashlib\n", + "from lxml.etree import HTML\n", + "import traceback\n", + "from datetime import datetime\n", + "import uuid\n", + "\n", + "# 配置\n", + "config = {\n", + " \"spark_conf_name\": \"spark_4\",\n", + " \"skip_success_check\": True,\n", + " \"spark.yarn.queue\": \"pipeline.clean\",\n", + " \"spark.dynamicAllocation.maxExecutors\": 2000, # 控制1万并发\n", + " # \"spark.executor.memory\": \"40g\",\n", + " # \"spark.executor.memoryOverhead\": \"20g\", # 增加到40GB\n", + " # \"spark.speculation\": \"true\", # 启用推测执行\n", + " # \"maxRecordsPerFile\": 200000, # 增加每文件记录数以减少总文件数\n", + " \"output_compression\": \"gz\",\n", + " \"skip_output_version\": True,\n", + " \"skip_output_check\": True,\n", + " \"spark.sql.shuffle.partitions\": \"20000\",\n", + " \"spark.default.parallelism\": \"20000\",\n", + " \"spark.network.timeout\": \"1200s\", # 网络超时\n", + " \"spark.broadcast.timeout\": \"1800s\", # 增加广播超时\n", + " \"spark.broadcast.compress\": \"true\", # 确保广播压缩\n", + " \"spark.task.maxFailures\": 8, \n", + "}\n", + "\n", + "from pyspark.sql.types import StructType, StructField, StringType\n", + "import re\n", + "import hashlib\n", + "from lxml.etree import HTML\n", + "\n", + "MAX_OUTPUT_ROW_SIZE = 1024 * 1024 * 1024 * 1.5\n", + "# DUMPS = [\n", + "# \"20230801\",\n", + "# ]\n", + "\n", + "# DUMPS = [line.strip() for line in open('pjcc_list.txt')]\n", + "# print(f\"加载了 {len(DUMPS)} 个dumps\")\n", + "\n", + "ERROR_PATH = 's3://qa-huawei/chupei/cc-domain-centric-store/error_logs/'\n", + "CC_WARC = 's3://cn-common-crawl/jsonl/'\n", + "output_path = \"s3://web-parse-hw60p/PJCC-dedup/hash/v5/\"\n", + "spark = new_spark_session(\"cc_dumps.dedup.fir\", config)\n", + "sc = spark.sparkContext" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "2", + "metadata": {}, + "outputs": [], + "source": [ + "# 查看样例和数量\n", + "print(f\"warc_paths 总数量: {len(warc_paths)}\")\n", + "if warc_paths:\n", + " print(f\"第一个文件路径示例: {warc_paths[0]}\")\n", + " print(f\"最后一个文件路径示例: {warc_paths[-1]}\")\n", + "else:\n", + " print(\"warc_paths 为空\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "3", + "metadata": {}, + "outputs": [], + "source": [ + "from pyspark.sql import Row\n", + "from xinghe.spark import *\n", + "from app.common.json_util import *\n", + "from xinghe.s3 import *\n", + "from pyspark.sql.types import StructType, StructField, StringType\n", + "import re\n", + "import hashlib\n", + "from lxml.etree import HTML\n", + "import traceback\n", + "from datetime import datetime\n", + "import uuid\n", + "\n", + "def html_to_content(html_str: str, url: str) -> str:\n", + " if html_str.strip() and isinstance(html_str,str):\n", + " html_str = re.sub(r'<\\?[^>]*\\?>', '', html_str.strip())\n", + " try:\n", + " html_etree = HTML(html_str)\n", + " except:\n", + " return None\n", + " if html_etree:\n", + " for element in html_etree.xpath('//*[self::script or self::style]'):\n", + " element.getparent().remove(element)\n", + " text = ''.join(html_etree.xpath(\"//text()\"))\n", + " cleaned_text = re.sub(r'[^\\w\\s]', '', text, flags=re.UNICODE)\n", + " cleaned_text = re.sub(r'\\s+', '', cleaned_text).strip()\n", + " return sha256_hash(cleaned_text)\n", + "\n", + "def sha256_hash(string):\n", + " return hashlib.sha256(string.encode()).hexdigest()\n", + " \n", + "# def parse_path_to_html(iter):\n", + "# seen = set()\n", + "# for fpath in iter:\n", + "# for zz in read_s3_rows(fpath, use_stream=True):\n", + "# detail_datas = json_loads(zz.value)\n", + "# hash_html = html_to_content(detail_datas.get(\"html\"), detail_datas[\"url\"]) if detail_datas.get(\"html\", \"\") else None\n", + "# if hash_html and hash_html not in seen:\n", + "# seen.add(hash_html)\n", + "# line = {\n", + "# \"sub_path\": fpath.split('/')[4],\n", + "# \"hash_html\": hash_html,\n", + "# \"track_id\": detail_datas[\"track_id\"],\n", + "# }\n", + "# yield Row(**{\"value\": json_dumps(line)})\n", + "\n", + "# 异常日志\n", + "def get_s3_doctor(target_theme):\n", + " partition_id = str(uuid.uuid4())\n", + " current_time = datetime.now().strftime(\"%Y%m%d\")\n", + " error_log_path = f\"{ERROR_PATH}{target_theme}/{current_time}/{partition_id}.jsonl\"\n", + " s3_doc_writer = S3DocWriter(path=error_log_path)\n", + " return s3_doc_writer\n", + "\n", + "def parse_path_to_html(iter):\n", + " seen = set() # 保持原有的 seen 逻辑\n", + " \n", + " # 初始化错误日志写入器\n", + " s3_doc_writer = get_s3_doctor(\"dedup_fir\")\n", + " error_info = None # 错误信息初始化\n", + " \n", + " for fpath in iter: \n", + " try:\n", + " # 读取文件并处理\n", + " for zz in read_s3_rows(fpath, use_stream=True):\n", + " try:\n", + " detail_datas = json_loads(zz.value)\n", + " # 安全地获取字段，提供默认值\n", + " html_content = detail_datas.get(\"html\", \"\")\n", + " url = detail_datas.get(\"url\", \"\")\n", + " track_id = detail_datas.get(\"track_id\", \"\")\n", + " \n", + " hash_html = html_to_content(html_content, url) if html_content else None\n", + " if hash_html and hash_html not in seen: # 保持原有的去重逻辑\n", + " seen.add(hash_html)\n", + " line = {\n", + " \"sub_path\": fpath.split('/')[4],\n", + " \"hash_html\": hash_html,\n", + " \"track_id\": track_id,\n", + " \"file_path\": fpath,\n", + " }\n", + " yield Row(**{\"value\": json_dumps(line)})\n", + " \n", + " except Exception as e:\n", + " # 记录数据解析错误\n", + " error_info = {\n", + " \"error_type\": type(e).__name__,\n", + " \"error_message\": str(e),\n", + " \"traceback\": traceback.format_exc(),\n", + " \"input_data\": zz.value if hasattr(zz, 'value') else str(zz),\n", + " \"file_path\": fpath,\n", + " \"timestamp\": datetime.now().isoformat()\n", + " }\n", + " s3_doc_writer.write(error_info)\n", + " continue\n", + " \n", + " except Exception as e:\n", + " # 记录文件读取错误\n", + " error_info = {\n", + " \"error_type\": type(e).__name__,\n", + " \"error_message\": str(e),\n", + " \"traceback\": traceback.format_exc(),\n", + " \"input_data\": \"N/A\",\n", + " \"file_path\": fpath,\n", + " \"timestamp\": datetime.now().isoformat()\n", + " }\n", + " s3_doc_writer.write(error_info)\n", + " continue\n", + " \n", + " if error_info:\n", + " # 确保所有错误日志都被写入\n", + " s3_doc_writer.flush()\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "4", + "metadata": {}, + "outputs": [], + "source": [ + "# mapPartitions 对 warc path 并行解析数据\n", + "schema = StructType([\n", + " StructField(\"value\", StringType(), True),\n", + "])\n", + "page_content = sc.parallelize(warc_paths, len(warc_paths))\n", + "dump_html_df = page_content.mapPartitions(parse_path_to_html).toDF(schema)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "5", + "metadata": {}, + "outputs": [], + "source": [ + "config[\"skip_output_version\"] = True\n", + "config[\"output_compression\"] = \"gz\"\n", + "write_any_path(dump_html_df, output_path, config)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "6", + "metadata": {}, + "outputs": [], + "source": [] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3.10 (ipykernel)", + "language": "python", + "name": "python3.10" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.10.9" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/jupyter/pjcc-prededup/cc_dedup_sec.ipynb b/jupyter/pjcc-prededup/cc_dedup_sec.ipynb new file mode 100644 index 00000000..a2c07f2c --- /dev/null +++ b/jupyter/pjcc-prededup/cc_dedup_sec.ipynb @@ -0,0 +1,421 @@ +{ + "cells": [ + { + "cell_type": "code", + "execution_count": null, + "id": "0", + "metadata": {}, + "outputs": [], + "source": [ + "\"\"\"\n", + "简单的hash_html去重脚本\n", + "\n", + "使用方法：\n", + "1. 直接调用函数：\n", + " hash_dedup_simple(spark, input_base_path, output_base_path, config)\n", + "\n", + "2. 参数说明：\n", + " - input_base_path: 输入数据路径\n", + " - output_base_path: 输出路径前缀\n", + " - config: xinghe配置对象\n", + "\n", + "3. 输出：\n", + " - 对hash_html字段进行全量去重\n", + " - 提取file_path后缀（如20241109/xxx.jsonl.gz），拼接output_base_path\n", + " - 使用S3DocWriter在partition内独立写文件，避免collect操作\n", + "\n", + "4. 数据格式：\n", + " 输入: value包含{\"sub_path\", \"hash_html\", \"track_id\", \"file_path\"}\n", + " 输出: 按file_path后缀分组保存到output_base_path/后缀路径\n", + "\"\"\"\n", + "\n", + "from pyspark.sql import SparkSession\n", + "from pyspark.sql.functions import *\n", + "from pyspark.sql.types import *\n", + "from xinghe.spark import *\n", + "from xinghe.s3 import *\n", + "import json\n", + "import uuid\n", + "import traceback\n", + "from datetime import datetime\n", + "\n", + "# 错误日志路径\n", + "ERROR_PATH = \"s3://qa-huawei/chupei/cc-domain-centric-store/error_logs/\"\n", + "\n", + "# 异常日志\n", + "def get_s3_doctor(target_theme):\n", + " partition_id = str(uuid.uuid4())\n", + " current_time = datetime.now().strftime(\"%Y%m%d\")\n", + " error_log_path = f\"{ERROR_PATH}{target_theme}/{current_time}/{partition_id}.jsonl\"\n", + " s3_doc_writer = S3DocWriter(path=error_log_path)\n", + " return s3_doc_writer\n", + "\n", + "\n", + "def write_partition_to_s3(iterator, output_base_path):\n", + " \"\"\"\n", + " 在partition内使用S3DocWriter写文件\n", + " \"\"\" \n", + " # 初始化错误日志写入器\n", + " s3_doc_writer = get_s3_doctor(\"dedup_sec\")\n", + " total_records = 0\n", + " \n", + " try:\n", + " # 按output_path分组数据\n", + " file_groups = {}\n", + " for row in iterator:\n", + " output_path = row.output_path\n", + " if output_path not in file_groups:\n", + " file_groups[output_path] = []\n", + " file_groups[output_path].append(row.output_value)\n", + " total_records += 1\n", + " \n", + " # 为每个文件写入数据\n", + " for output_path, values in file_groups.items():\n", + " try:\n", + " writer = S3DocWriter(output_path)\n", + " for value in values:\n", + " # 将JSON字符串转换回字典，S3DocWriter会自动添加换行符\n", + " data_dict = json.loads(value)\n", + " writer.write(data_dict)\n", + " writer.flush()\n", + " print(f\"成功写入文件: {output_path}, 记录数: {len(values)}\")\n", + " except Exception as e:\n", + " print(f\"写入文件失败: {output_path}, 错误: {str(e)}\")\n", + " \n", + " except Exception as e:\n", + " # 外层异常捕获，记录到错误日志\n", + " error_info = {\n", + " \"error_type\": type(e).__name__,\n", + " \"error_message\": str(e),\n", + " \"traceback\": traceback.format_exc(),\n", + " \"input_data\": \"partition_processing_error\",\n", + " \"stage\": \"write_partition_to_s3\",\n", + " \"timestamp\": datetime.now().isoformat()\n", + " }\n", + " s3_doc_writer.write(error_info)\n", + " s3_doc_writer.flush()\n", + " print(f\"分区处理失败: {str(e)}\")\n", + " \n", + " return iter([total_records]) # 返回该分区处理的记录数\n", + "\n", + "def hash_dedup_simple(spark, input_base_path, output_base_path, config):\n", + " \"\"\"\n", + " 简单的hash_html去重处理\n", + " \n", + " Args:\n", + " spark: SparkSession\n", + " input_base_path: 输入数据路径\n", + " output_base_path: 输出路径前缀\n", + " config: xinghe配置对象\n", + " \"\"\"\n", + " print(f\"开始处理hash_html去重...\")\n", + " print(f\"输入路径: {input_base_path}\")\n", + " print(f\"输出路径前缀: {output_base_path}\")\n", + " \n", + " # 使用read_any_path读取数据\n", + " input_df = read_any_path(spark, input_base_path, config)\n", + " \n", + " # 解析JSON数据\n", + " parsed_df = input_df.select(\n", + " from_json(col(\"value\"), StructType([\n", + " StructField(\"sub_path\", StringType()),\n", + " StructField(\"hash_html\", StringType()),\n", + " StructField(\"track_id\", StringType()),\n", + " StructField(\"file_path\", StringType())\n", + " ])).alias(\"data\")\n", + " ).select(\n", + " col(\"data.*\")\n", + " ).filter(col(\"hash_html\").isNotNull())\n", + " \n", + " print(\"开始全量去重...\")\n", + " # 全量去重\n", + " deduped_df = parsed_df.dropDuplicates([\"hash_html\"])\n", + " \n", + " # 构造输出路径和数据\n", + " result_df = deduped_df.withColumn(\n", + " \"output_path\",\n", + " concat(\n", + " lit(output_base_path + \"/\"),\n", + " regexp_extract(col(\"file_path\"), r\".*/([^/]+/[^/]+\\.jsonl(?:\\.gz)?)$\", 1)\n", + " )\n", + " ).withColumn(\n", + " \"output_value\",\n", + " to_json(struct(\"sub_path\", \"hash_html\", \"track_id\", \"file_path\"))\n", + " )\n", + " \n", + " print(\"开始写入文件...\")\n", + " # 使用mapPartitions在每个partition内独立写文件，避免collect\n", + " # 按output_path分区，确保同一output_path的数据在同一partition\n", + " total_written_records = result_df.repartition(col(\"output_path\")) \\\n", + " .rdd \\\n", + " .mapPartitions(lambda iterator: write_partition_to_s3(iterator, output_base_path)) \\\n", + " .sum() # 对所有分区的记录数求和\n", + " \n", + " print(f\"处理完成! 写入的总记录数: {total_written_records}\")\n", + " return deduped_df\n", + "\n", + "def hash_dedup_incremental(spark, input_base_path, output_base_path, existing_base_path, config):\n", + " \"\"\"\n", + " 增量hash_html去重处理\n", + " \n", + " Args:\n", + " spark: SparkSession\n", + " input_base_path: 新增数据路径 (如 v6)\n", + " output_base_path: 输出路径前缀 (如 v2)\n", + " existing_base_path: 已存在的去重数据路径 (如 v2，用于读取已有hash_html)\n", + " config: xinghe配置对象\n", + " \"\"\"\n", + " print(f\"开始处理增量hash_html去重...\")\n", + " print(f\"新增数据路径: {input_base_path}\")\n", + " print(f\"已存在数据路径: {existing_base_path}\")\n", + " print(f\"输出路径前缀: {output_base_path}\")\n", + " \n", + " # 1. 读取新增数据\n", + " print(\"读取新增数据...\")\n", + " new_input_df = read_any_path(spark, input_base_path, config)\n", + " \n", + " # 解析新增数据\n", + " new_parsed_df = new_input_df.select(\n", + " from_json(col(\"value\"), StructType([\n", + " StructField(\"sub_path\", StringType()),\n", + " StructField(\"hash_html\", StringType()),\n", + " StructField(\"track_id\", StringType()),\n", + " StructField(\"file_path\", StringType())\n", + " ])).alias(\"data\")\n", + " ).select(\n", + " col(\"data.*\")\n", + " ).filter(col(\"hash_html\").isNotNull())\n", + " \n", + " print(\"新增数据读取完成\")\n", + " \n", + " # 2. 高效去重策略：避免读取全量已存在数据\n", + " print(\"开始高效增量去重...\")\n", + " \n", + " # 策略：直接对新数据进行去重，然后使用left_anti join过滤\n", + " # 先对新数据内部去重，减少需要join的数据量\n", + " print(\"对新数据进行内部去重...\")\n", + " new_internal_deduped = new_parsed_df.dropDuplicates([\"hash_html\"])\n", + " print(\"新数据内部去重完成\")\n", + " \n", + " # 只读取已存在数据的hash_html字段，避免读取完整记录\n", + " print(\"读取已存在数据的hash_html...\")\n", + " try:\n", + " existing_df = read_any_path(spark, existing_base_path, config)\n", + " # 直接提取hash_html，避免解析完整JSON\n", + " existing_hashes = existing_df.select(\n", + " get_json_object(col(\"value\"), \"$.hash_html\").alias(\"hash_html\")\n", + " ).filter(col(\"hash_html\").isNotNull())\n", + " \n", + " print(\"已存在hash_html读取完成\")\n", + " \n", + " # 使用分区优化的join，确保相同hash在同一分区\n", + " print(\"执行高效去重join...\")\n", + " new_unique_df = new_internal_deduped.repartition(col(\"hash_html\")) \\\n", + " .join(existing_hashes.repartition(col(\"hash_html\")), \n", + " [\"hash_html\"], \"left_anti\")\n", + " \n", + " except Exception as e:\n", + " print(f\"读取已存在数据失败（可能是首次运行）: {str(e)}\")\n", + " print(\"请使用 hash_dedup_simple 函数进行首次全量去重\")\n", + " raise e\n", + " \n", + " # 4. 过滤完成\n", + " print(\"高效去重完成...\")\n", + " \n", + " print(\"重复数据过滤完成\")\n", + " \n", + " # 5. 构造输出路径和数据（new_unique_df已经是去重后的数据）\n", + " result_df = new_unique_df.withColumn(\n", + " \"output_path\",\n", + " concat(\n", + " lit(output_base_path + \"/\"),\n", + " regexp_extract(col(\"file_path\"), r\".*/([^/]+/[^/]+\\.jsonl(?:\\.gz)?)$\", 1)\n", + " )\n", + " ).withColumn(\n", + " \"output_value\",\n", + " to_json(struct(\"sub_path\", \"hash_html\", \"track_id\", \"file_path\"))\n", + " )\n", + " \n", + " print(\"开始写入新增数据...\")\n", + " # 使用mapPartitions在每个partition内独立写文件，和hash_dedup_simple一致\n", + " total_written_records = result_df.repartition(col(\"output_path\")) \\\n", + " .rdd \\\n", + " .mapPartitions(lambda iterator: write_partition_to_s3(iterator, output_base_path)) \\\n", + " .sum() # 对所有分区的记录数求和\n", + " \n", + " print(f\"增量去重处理完成! 写入的总记录数: {total_written_records}\")\n", + " return new_unique_df\n", + "\n", + "\n", + "\n", + "# 使用示例\n", + "if __name__ == \"__main__\":\n", + " \n", + " # 配置\n", + " config = {\n", + " \"spark_conf_name\": \"spark_4\",\n", + " \"skip_success_check\": True,\n", + " \"spark.yarn.queue\": \"pipeline.clean\",\n", + " \"spark.dynamicAllocation.maxExecutors\": 1000, # 控制1万并发\n", + " \"spark.executor.memory\": \"80g\",\n", + " \"spark.executor.memoryOverhead\": \"40g\", # 增加到40GB\n", + " # \"spark.speculation\": \"true\", # 启用推测执行\n", + " # \"maxRecordsPerFile\": 200000, # 增加每文件记录数以减少总文件数\n", + " \"output_compression\": \"gz\",\n", + " \"skip_output_version\": True,\n", + " \"skip_output_check\": True,\n", + " \"spark.sql.shuffle.partitions\": \"10000\", # 减少分区数\n", + " \"spark.default.parallelism\": \"10000\",\n", + " \n", + " # Shuffle 优化配置\n", + " \"spark.shuffle.io.maxRetries\": \"10\", # 增加shuffle重试次数\n", + " \"spark.shuffle.io.retryWait\": \"30s\", # 重试等待时间\n", + " \"spark.shuffle.compress\": \"true\", # 启用shuffle压缩\n", + " \"spark.shuffle.spill.compress\": \"true\", # 启用spill压缩\n", + " \n", + " # 网络和超时配置\n", + " \"spark.network.timeout\": \"3600s\", # 进一步增加网络超时\n", + " \"spark.broadcast.timeout\": \"3600s\", \n", + " \"spark.broadcast.compress\": \"true\",\n", + " \"spark.rpc.askTimeout\": \"3600s\", \n", + " \"spark.rpc.lookupTimeout\": \"3600s\", \n", + " \"spark.storage.blockManagerSlaveTimeoutMs\": \"3600000\",\n", + " \n", + " }\n", + " spark = new_spark_session(\"cc_dumps.dedup.sec\", config)\n", + " sc = spark.sparkContext\n", + " sc.setLogLevel(\"ERROR\")\n", + " sc\n", + " \n", + " # 示例用法\n", + " \n", + " # 方案1: 全量去重（首次运行）\n", + " # input_base_path = \"s3://web-parse-hw60p/PJCC-dedup/hash/v5\"\n", + " # output_base_path = \"s3://web-parse-hw60p/PJCC-dedup/hash-dedup/v2\"\n", + " # result = hash_dedup_simple(spark, input_base_path, output_base_path, config)\n", + " \n", + " # 方案2: 增量去重（后续运行）\n", + " input_base_path = \"s3://web-parse-hw60p/PJCC-dedup/hash/v6\" # 新增数据\n", + " output_base_path = \"s3://web-parse-hw60p/PJCC-dedup/hash-dedup/v2\" # 输出路径\n", + " existing_base_path = \"s3://web-parse-hw60p/PJCC-dedup/hash-dedup/v2\" # 已存在数据路径\n", + " \n", + " # 执行增量去重\n", + " result = hash_dedup_incremental(spark, input_base_path, output_base_path, existing_base_path, config)\n", + " \n", + " spark.stop() " + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "1", + "metadata": {}, + "outputs": [], + "source": [ + "# Spark配置\n", + "config = {\n", + " \"spark_conf_name\": \"spark_4\",\n", + " \"skip_success_check\": True,\n", + " \"spark.yarn.queue\": \"pipeline.clean\",\n", + " \"spark.dynamicAllocation.maxExecutors\": 2000, # 控制1万并发\n", + " \"output_compression\": \"gz\",\n", + " \"skip_output_version\": True,\n", + " \"skip_output_check\": True,\n", + " \"spark.sql.shuffle.partitions\": \"20000\",\n", + " \"spark.default.parallelism\": \"20000\",\n", + " \"spark.network.timeout\": \"1200s\", # 网络超时\n", + " \"spark.broadcast.timeout\": \"1800s\", # 增加广播超时\n", + " \"spark.broadcast.compress\": \"true\", # 确保广播压缩\n", + " \"spark.task.maxFailures\": 8,\n", + "}\n", + " \n", + "from xinghe.spark.session_ext import new_spark_session \n", + "from xinghe.spark.read_ext import read_any_path\n", + "spark = new_spark_session(\"extract_unique_file_paths\", config)\n", + "sc = spark.sparkContext\n", + "sc.setLogLevel(\"ERROR\")\n", + "input_path = \"s3://web-parse-hw60p/PJCC-dedup/hash-dedup/v2/\"\n", + "input_df = read_any_path(spark, input_path, config)\n", + " \n", + "total_records = input_df.count()\n", + "print(f\"总记录数: {total_records:,}\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "2", + "metadata": {}, + "outputs": [], + "source": [ + "from pyspark.sql import SparkSession\n", + "from pyspark.sql.functions import *\n", + "from pyspark.sql.types import *\n", + "from xinghe.spark import *\n", + "from xinghe.s3 import *\n", + "import json\n", + "import uuid\n", + "import traceback\n", + "from datetime import datetime\n", + "\n", + "input_base_path = \"s3://web-parse-hw60p/PJCC-dedup/hash/v5/20230801\"\n", + "input_df = read_any_path(spark, input_base_path, config)\n", + "input_df.count()" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "3", + "metadata": {}, + "outputs": [], + "source": [ + "input_base_path = \"s3://web-parse-hw60p/PJCC-dedup/hash_dedup/v1/20230801\"\n", + "input_df = read_any_path(spark, input_base_path, config)\n", + "input_df.count()" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "4", + "metadata": {}, + "outputs": [], + "source": [ + "files = list(list_s3_objects('s3://web-parse-hw60p/PJCC-dedup/hash/v5/20230801/', is_prefix=True))\n", + "print(len(files))\n", + "files = list(list_s3_objects('s3://web-parse-hw60p/PJCC-dedup/hash_dedup/v1/20230801/', is_prefix=True))\n", + "print(len(files))" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "5", + "metadata": {}, + "outputs": [], + "source": [] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3.10 (ipykernel)", + "language": "python", + "name": "python3.10" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.10.9" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/jupyter/pjcc-prededup/cc_dedup_thr.ipynb b/jupyter/pjcc-prededup/cc_dedup_thr.ipynb new file mode 100644 index 00000000..fcaf0076 --- /dev/null +++ b/jupyter/pjcc-prededup/cc_dedup_thr.ipynb @@ -0,0 +1,348 @@ +{ + "cells": [ + { + "cell_type": "code", + "execution_count": null, + "id": "0", + "metadata": {}, + "outputs": [], + "source": [ + "\"\"\"\n", + "根据hash_html去重后数据，进行html去重脚本\n", + "\n", + "使用方法：\n", + "1. 直接调用函数：\n", + " html_dedup(spark, input_hash_base_path, input_html_base_path, output_html_base_path, output_hash_with_domain_base_path, config, index_path)\n", + "\n", + "2. 参数说明：\n", + " - input_hash_base_path: 输入hash_html去重后的数据路径\n", + " - input_html_base_path: 输入原始html数据路径\n", + " - output_html_base_path: 输出html去重后的数据路径\n", + " - output_hash_with_domain_base_path: 输出hash+domain数据路径\n", + " - config: xinghe配置对象\n", + " - index_path: 文件列表路径，用于获取需要处理的文件后缀\n", + "\n", + "3. 处理逻辑：\n", + " - 从index_path读取文件列表，提取文件后缀进行并行处理\n", + " - 对每个文件后缀，分别读取hash去重数据和原始html数据\n", + " - 通过track_id字段进行inner join，保留在hash去重结果中的html数据\n", + " - 从URL提取domain和domain_hash_id字段\n", + " - 生成两个输出：HTML去重数据和Hash+Domain数据\n", + "\n", + "4. 文件路径示例：\n", + " - Hash去重文件: s3://web-parse-hw60p/PJCC-dedup/hash-dedup/v1/20230801/xxx.jsonl.gz\n", + " - 原始HTML文件: s3://cn-common-crawl/jsonl/20230801/xxx.jsonl.gz\n", + " - 输出HTML文件: s3://web-parse-hw60p/PJCC-dedup/html-dedup/v1/20230801/xxx.jsonl.gz\n", + " - 输出Hash+Domain文件: s3://web-parse-hw60p/PJCC-dedup/hash-dedup-with-domain/v1/20230801/xxx.jsonl.gz\n", + "\n", + "5. 数据格式：\n", + " - Hash去重数据: {\"sub_path\", \"hash_html\", \"track_id\", \"file_path\"}\n", + " - 原始HTML数据: {\"sub_path\", \"html\", \"track_id\", \"file_path\", \"url\", ...}\n", + " - 输出HTML数据: 原始HTML数据 + domain相关字段\n", + " - 输出Hash+Domain数据: Hash去重数据 + {\"url\", \"domain\", \"domain_hash_id\"}\n", + "\"\"\"\n", + "\n", + "from pyspark.sql import SparkSession\n", + "from pyspark.sql.functions import *\n", + "from pyspark.sql.types import *\n", + "from xinghe.spark import *\n", + "from xinghe.s3 import *\n", + "import json\n", + "import uuid\n", + "import traceback\n", + "from datetime import datetime\n", + "from urllib.parse import urlparse\n", + "import xxhash\n", + "\n", + "# 错误日志路径\n", + "ERROR_PATH = \"s3://qa-huawei/chupei/cc-domain-centric-store/error_logs/\"\n", + "\n", + "# 异常日志\n", + "def get_s3_doctor(target_theme):\n", + " partition_id = str(uuid.uuid4())\n", + " current_time = datetime.now().strftime(\"%Y%m%d\")\n", + " error_log_path = f\"{ERROR_PATH}{target_theme}/{current_time}/{partition_id}.jsonl\"\n", + " s3_doc_writer = S3DocWriter(path=error_log_path)\n", + " return s3_doc_writer\n", + "\n", + "# 定义提取domain的UDF\n", + "def extract_domain(url):\n", + " if url is None:\n", + " return None\n", + " try:\n", + " hostname = urlparse(url).hostname\n", + " return hostname.lower() if hostname else None\n", + " except Exception as e:\n", + " return None\n", + "\n", + "# 定义计算domain_hash_id的UDF\n", + "HASH_COUNT = 10000\n", + "def compute_domain_hash(domain):\n", + " if domain is None:\n", + " return None\n", + " return xxhash.xxh64_intdigest(domain) % HASH_COUNT\n", + "\n", + "def process_file_pair(file_suffix, input_hash_base_path, input_html_base_path, output_html_base_path, output_hash_with_domain_base_path):\n", + " \"\"\"\n", + " 处理一对文件的html去重 - 在partition内使用直接文件操作\n", + " \"\"\"\n", + " \n", + " # 初始化错误日志写入器\n", + " s3_doc_writer = get_s3_doctor(\"dedup_thr\")\n", + " \n", + " try:\n", + " # 构造文件路径\n", + " hash_file_path = f\"{input_hash_base_path}/{file_suffix}\"\n", + " html_file_path = f\"{input_html_base_path}/{file_suffix}\"\n", + " output_file_path = f\"{output_html_base_path}/{file_suffix}\"\n", + " output_hash_with_domain_file_path = f\"{output_hash_with_domain_base_path}/{file_suffix}\"\n", + " \n", + " print(f\"开始处理文件对:\")\n", + " print(f\" Hash文件: {hash_file_path}\")\n", + " print(f\" HTML文件: {html_file_path}\")\n", + " print(f\" 输出HTML文件: {output_file_path}\")\n", + " print(f\" 输出Hash+Domain文件: {output_hash_with_domain_file_path}\")\n", + " \n", + " # 读取hash去重后的数据\n", + " hash_data_dict = {} # track_id -> hash_data\n", + " for zz in read_s3_rows(hash_file_path, use_stream=True):\n", + " try:\n", + " hash_detail = json.loads(zz.value)\n", + " track_id = hash_detail.get(\"track_id\")\n", + " if track_id:\n", + " hash_data_dict[track_id] = hash_detail\n", + " except Exception as e:\n", + " # 记录解析hash数据的错误\n", + " error_info = {\n", + " \"error_type\": type(e).__name__,\n", + " \"error_message\": str(e),\n", + " \"traceback\": traceback.format_exc(),\n", + " \"input_data\": f\"file_suffix: {file_suffix}, hash_file: {hash_file_path}, raw_data: {zz.value[:200] if hasattr(zz, 'value') else 'N/A'}\",\n", + " \"stage\": \"parse_hash_data\",\n", + " \"timestamp\": datetime.now().isoformat()\n", + " }\n", + " s3_doc_writer.write(error_info)\n", + " print(f\"解析hash数据失败: {str(e)}\")\n", + " continue\n", + " \n", + " print(f\"读取hash数据: {len(hash_data_dict)} 条记录\")\n", + " \n", + " # 初始化writer - 在循环外创建，避免重复创建\n", + " html_writer = S3DocWriter(output_file_path)\n", + " hash_domain_writer = S3DocWriter(output_hash_with_domain_file_path)\n", + " \n", + " # 读取原始html数据并进行join，同时写入结果\n", + " html_count = 0\n", + " hash_domain_count = 0\n", + " \n", + " for zz in read_s3_rows(html_file_path, use_stream=True):\n", + " try:\n", + " html_detail = json.loads(zz.value)\n", + " track_id = html_detail.get(\"track_id\")\n", + " \n", + " # 通过track_id进行inner join\n", + " if track_id and track_id in hash_data_dict:\n", + " # 处理URL提取domain\n", + " url = html_detail.get(\"url\")\n", + " domain = extract_domain(url)\n", + " domain_hash_id = compute_domain_hash(domain)\n", + " \n", + " # 构造HTML去重结果并立即写入\n", + " html_result = html_detail.copy()\n", + " html_result[\"domain\"] = domain\n", + " html_result[\"domain_hash_id\"] = domain_hash_id\n", + " html_writer.write(html_result)\n", + " html_count += 1\n", + " \n", + " # 构造Hash+Domain结果并立即写入\n", + " hash_result = hash_data_dict[track_id].copy()\n", + " # 去掉sub_path字段\n", + " if \"sub_path\" in hash_result:\n", + " del hash_result[\"sub_path\"]\n", + " hash_result[\"url\"] = url\n", + " hash_result[\"domain\"] = domain\n", + " hash_result[\"domain_hash_id\"] = domain_hash_id\n", + " hash_domain_writer.write(hash_result)\n", + " hash_domain_count += 1\n", + " \n", + " except Exception as e:\n", + " # 记录处理html数据的错误\n", + " error_info = {\n", + " \"error_type\": type(e).__name__,\n", + " \"error_message\": str(e),\n", + " \"traceback\": traceback.format_exc(),\n", + " \"input_data\": f\"file_suffix: {file_suffix}, html_file: {html_file_path}, raw_data: {zz.value[:200] if hasattr(zz, 'value') else 'N/A'}\",\n", + " \"stage\": \"process_html_data\",\n", + " \"timestamp\": datetime.now().isoformat()\n", + " }\n", + " s3_doc_writer.write(error_info)\n", + " print(f\"处理html数据失败: {str(e)}\")\n", + " continue\n", + " \n", + " # 确保所有数据都写入\n", + " html_writer.flush()\n", + " hash_domain_writer.flush()\n", + " \n", + " print(f\"Join结果: {html_count} 条记录\")\n", + " print(f\"成功写入HTML去重文件: {output_file_path}, 记录数: {html_count}\")\n", + " print(f\"成功写入Hash+Domain文件: {output_hash_with_domain_file_path}, 记录数: {hash_domain_count}\")\n", + " \n", + " return html_count\n", + " \n", + " except Exception as e:\n", + " # 记录错误日志\n", + " error_info = {\n", + " \"error_type\": type(e).__name__,\n", + " \"error_message\": str(e),\n", + " \"traceback\": traceback.format_exc(),\n", + " \"input_data\": f\"file_suffix: {file_suffix}\",\n", + " \"stage\": \"process_file_pair\",\n", + " \"timestamp\": datetime.now().isoformat()\n", + " }\n", + " s3_doc_writer.write(error_info)\n", + " s3_doc_writer.flush()\n", + " print(f\"文件处理失败: {file_suffix}, 错误: {str(e)}\")\n", + " return 0\n", + "\n", + "def html_dedup(spark, input_hash_base_path, input_html_base_path, output_html_base_path, output_hash_with_domain_base_path, config, index_path):\n", + " \"\"\"\n", + " 根据hash_html去重后数据，进行html去重\n", + " \n", + " Args:\n", + " spark: SparkSession\n", + " input_hash_base_path: 输入hash_html去重后的数据路径\n", + " input_html_base_path: 输入html数据路径\n", + " output_html_base_path: 输出html去重后的数据路径\n", + " output_hash_with_domain_base_path: 输出hash+domain数据路径\n", + " config: xinghe配置对象\n", + " index_path: 文件列表路径\n", + " \"\"\"\n", + " print(f\"开始处理html去重...\")\n", + " print(f\"Hash去重数据路径: {input_hash_base_path}\")\n", + " print(f\"HTML数据路径: {input_html_base_path}\")\n", + " print(f\"输出HTML路径: {output_html_base_path}\")\n", + " print(f\"输出Hash+Domain路径: {output_hash_with_domain_base_path}\")\n", + " print(f\"文件列表路径: {index_path}\")\n", + " \n", + " # 读取文件列表\n", + " index_df = read_any_path(spark, index_path, config)\n", + " paths = index_df.rdd.map(lambda row: row.value).collect()\n", + " \n", + " print(f\"总文件数: {len(paths)}\")\n", + " \n", + " # 并行处理文件\n", + " paths_rdd = spark.sparkContext.parallelize(paths)\n", + " \n", + " # 使用mapPartitions进行批量处理\n", + " def process_partition(file_paths):\n", + " results = []\n", + " for file_path in file_paths:\n", + " # 提取文件后缀\n", + " file_suffix = file_path.split('/')[-2] + '/' + file_path.split('/')[-1]\n", + " result_count = process_file_pair(\n", + " file_suffix, \n", + " input_hash_base_path, \n", + " input_html_base_path, \n", + " output_html_base_path, \n", + " output_hash_with_domain_base_path\n", + " )\n", + " results.append((file_suffix, result_count))\n", + " return results\n", + " \n", + " # 执行处理\n", + " paths_rdd.mapPartitions(process_partition).count() # 触发执行\n", + " \n", + " print(f\"处理完成!\")\n", + " print(f\"总文件数: {len(paths)}\")\n", + " \n", + " return True\n", + "\n", + "\n", + "# 使用示例\n", + "if __name__ == \"__main__\":\n", + " \n", + " # 配置\n", + " config = {\n", + " \"spark_conf_name\": \"spark_4\",\n", + " \"skip_success_check\": True,\n", + " \"spark.yarn.queue\": \"pipeline.clean\",\n", + " \"spark.dynamicAllocation.maxExecutors\": 2000, # 控制1万并发\n", + " # \"spark.executor.memory\": \"80g\",\n", + " # \"spark.executor.memoryOverhead\": \"40g\", # 增加到40GB\n", + " # \"spark.speculation\": \"true\", # 启用推测执行\n", + " # \"maxRecordsPerFile\": 200000, # 增加每文件记录数以减少总文件数\n", + " \"output_compression\": \"gz\",\n", + " \"skip_output_version\": True,\n", + " \"skip_output_check\": True,\n", + " \"spark.sql.shuffle.partitions\": \"20000\", # 减少分区数\n", + " \"spark.default.parallelism\": \"20000\",\n", + " \n", + " # Shuffle 优化配置\n", + " \"spark.shuffle.io.maxRetries\": \"10\", # 增加shuffle重试次数\n", + " \"spark.shuffle.io.retryWait\": \"30s\", # 重试等待时间\n", + " \"spark.shuffle.compress\": \"true\", # 启用shuffle压缩\n", + " \"spark.shuffle.spill.compress\": \"true\", # 启用spill压缩\n", + " \n", + " # 网络和超时配置\n", + " \"spark.network.timeout\": \"3600s\", # 进一步增加网络超时\n", + " \"spark.broadcast.timeout\": \"3600s\", \n", + " \"spark.broadcast.compress\": \"true\",\n", + " \"spark.rpc.askTimeout\": \"3600s\", \n", + " \"spark.rpc.lookupTimeout\": \"3600s\", \n", + " \"spark.storage.blockManagerSlaveTimeoutMs\": \"3600000\",\n", + " \n", + " }\n", + " spark = new_spark_session(\"cc_dumps.dedup.thr\", config)\n", + " sc = spark.sparkContext\n", + " sc.setLogLevel(\"ERROR\")\n", + " sc\n", + " \n", + " # 示例用法\n", + " input_hash_base_path = \"s3://web-parse-hw60p/PJCC-dedup/hash-dedup/v2\"\n", + " input_html_base_path = \"s3://cn-common-crawl/jsonl\"\n", + " output_html_base_path = \"s3://web-parse-hw60p/PJCC-dedup/html-dedup/v1\"\n", + " output_hash_with_domain_base_path = \"s3://web-parse-hw60p/PJCC-dedup/hash-dedup-with-domain/v1\"\n", + " index_path = \"s3://qa-huawei/chupei/cc-domain-centric-store/PJCC/FILELIST/pjcc-hash-dedup-v2.txt\"\n", + " # test config\n", + " # output_html_base_path = \"s3://qa-huawei/chupei/cc-domain-centric-store/PJCC-dedup/html-dedup/v1\"\n", + " # output_hash_with_domain_base_path = \"s3://qa-huawei/chupei/cc-domain-centric-store/PJCC-dedup/hash-dedup-with-domain/v1\"\n", + " # index_path = \"s3://qa-huawei/chupei/cc-domain-centric-store/PJCC/FILELIST/pjcc-hash-dedup-test100.txt\"\n", + " \n", + " \n", + " # 执行去重\n", + " result = html_dedup(spark, input_hash_base_path, input_html_base_path, output_html_base_path, output_hash_with_domain_base_path, config, index_path)\n", + " \n", + " spark.stop() " + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "1", + "metadata": {}, + "outputs": [], + "source": [] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3.10 (ipykernel)", + "language": "python", + "name": "python3.10" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.10.9" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +}