feat: add pjcc prededup

.gitignore

@@ -11,7 +11,6 @@ __pycache__
 *.pyc
 .vscode
 debug/
-*.ipynb
 .idea
 .python-version
 

jupyter/domain_clustering/libs/domain.py

@@ -7,7 +7,7 @@
 def compute_domain_hash(domain, hash_count) -> int:
     if domain is None:
         return None
-    return xxhash.xxh64_intdigest(domain) % hash_count
+    return xxhash.xxh64_intdigest(domain.encode('utf-8')) % hash_count
 
 
 # 定义提取domain的UDF

jupyter/domain_clustering/pipeline/cc_domain_index_gen.py

@@ -180,7 +180,7 @@ def process_domain_records_file(_iter):
                     domain_hash_id = detail_data.get("domain_hash_id")
                     # 如果domain_hash_id为空，则计算
                     if domain_hash_id is None:
-                        domain_hash_id = xxhash.xxh64_intdigest(domain) % HASH_COUNT
+                        domain_hash_id = xxhash.xxh64_intdigest(domain.encode('utf-8')) % HASH_COUNT
                     offset, length = map(int, row.loc.split("bytes=")[-1].split(",")) if "bytes=" in row.loc else (0, len(row.value))
 
                     # 如果是新域名，先输出前一个域名的记录

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
+}