cc-index-table/bulk-url-lookups-by-table-joins.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "iraqi-shanghai",
   "metadata": {},
   "source": [
    "# Table Joins to Look Up Large Lists of URLs in the Common Crawl\n",
    "\n",
    "\n",
    "## A Short Introduction to URL indexes and SURT URLs\n",
    "\n",
    "### The CDX Index\n",
    "\n",
    "The Common Crawl CDX index [index.commoncrawl.org](https://index.commoncrawl.org/) allows to look up quickly whether a web page is contained in one of the monthly main crawls. Just a single request\n",
    "```\n",
    "curl -s 'https://index.commoncrawl.org/CC-MAIN-2021-10-index?url=https://commoncrawl.org/2021/02/january-2021-crawl-archive-now-available/&output=cdxj'\n",
    "```\n",
    "is sufficient to figure out that the announcement of the January crawl is included in the archives of the February crawl:\n",
    "```\n",
    "org,commoncrawl)/2021/02/january-2021-crawl-archive-now-available 20210303023607 {\"url\": \"https://commoncrawl.org/2021/02/january-2021-crawl-archive-now-available/\", \"mime\": \"text/html\", \"mime-detected\": \"text/html\", \"status\": \"200\", \"digest\": \"6JKL2JMU4HFJMZAH65BZZET653JJP53I\", \"length\": \"6922\", \"offset\": \"250975924\", \"filename\": \"crawl-data/CC-MAIN-2021-10/segments/1614178365186.46/warc/CC-MAIN-20210303012222-20210303042222-00595.warc.gz\", \"languages\": \"eng\", \"encoding\": \"UTF-8\"}\n",
    "```\n",
    "\n",
    "Every line in the index contains\n",
    "* the URL search key aka. SURT (\"Sort-friendly URI Reordering Transform\")\n",
    "* the capture time stamp\n",
    "* a JSON object containing the URL, metadata and the location of a capture in the crawl archives\n",
    "\n",
    "The so-called [zipnum sharded index](https://pywb.readthedocs.io/en/latest/manual/indexing.html?highlight=zipnum#zipnum-sharded-index) is sorted by the SURT URL key and split over 300 files. The index lines are gzip-compressed in blocks of 3000 lines. A secondary block index makes it possible to determine the block which should include a certain URL performing a quick binary search. Then the block is uncompressed to verify whether the URL is included, and if yes to extract the matched record(s).\n",
    "\n",
    "Because the host name in the SURT URL is reversed (`subdomain.example.org` becomes `org,example,subdomain`), queries for all subdomains of a domain are possible by doing a prefix search. Other normalizations make the SURT URL even more powerful for prefix look-ups: eg. sorting of URL query parameters, removal of the protocol (`https://`) or stripping off the leading `www.` in host names. These variations in the URL often lead to redirects or duplicated content. However, they are not safe to be normalized in a hard way because in few cases web servers may serve different page content for different URLs sharing the same SURT representation.\n",
    "\n",
    "\n",
    "### The Columnar Index\n",
    "\n",
    "The [columnar index](https://commoncrawl.org/2018/03/index-to-warc-files-and-urls-in-columnar-format/) has the following differences compared to the CDX index:\n",
    "- look-ups are not limited to the SURT URL, but are possible ony any provided column including content type (MIME), content languages and all URL parts (host name, registered domain, path, query)\n",
    "- the columnar structure is efficient and saves costs if only a subset of the columns is accessed\n",
    "- big data tools support SQL queries and aggregations ([Athena](https://aws.amazon.com/athena/)/[Presto](https://prestodb.io/), [Spark](https://spark.apache.org/), [Hive](https://hive.apache.org/))\n",
    "- however, usage of a tool able to query the [Parquet file format](https://parquet.apache.org/) is required\n",
    "\n",
    "\n",
    "### Index Partitioning\n",
    "\n",
    "For practical reasons - because 3 billion pages contribute to an index size of about 300 GiB, every monthly crawl is indexed separately. So, you would typically use a CDX client (eg. [cdx-toolkit](https://pypi.org/project/cdx-toolkit/)) which takes the work to iterate over a list of indexes for a given time window.\n",
    "\n",
    "Also the [columnar index](https://commoncrawl.org/2018/03/index-to-warc-files-and-urls-in-columnar-format/) stores every monthly crawl in a separate partition to alleviate the addition of a new crawl and also to target queries to individual crawls at zero computational cost.\n",
    "\n",
    "\n",
    "## Efficiently Looking Up Large Lists of URLs\n",
    "\n",
    "Back to our problem: how to look up a large list of URLs in the Common Crawl index...\n",
    "\n",
    "Unless the URLs share a common prefix, for example a domain name, the CDX index isn't the best data structure to do this. The task translates into a series of look ups and does not scale well - the number of queries is equal to the number of URLs multiplied by the number of requested monthly indexes.\n",
    "\n",
    "Similarly, runnning multiple queries over the columnar index isn't efficient. Instead we\n",
    "\n",
    "1. create a support table which holds the list of URLs we want to look up\n",
    "2. [Amazon Athena](https://aws.amazon.com/athena/) will do a table join to perform the intersection\n",
    "3. the result table holds an export of the record coordinates but can be also utilized to further analyze or filter the result.\n",
    "\n",
    "We use the SURT URLs to get a better recall without the need to artificially generate potential variants.\n",
    "\n",
    "\n",
    "### Variants and Further Optimizations\n",
    "\n",
    "- if possible try to further restrict the selection to perform the join on:\n",
    "  - few monthly crawls\n",
    "  - select on a crawl subset (`warc`: successful fetches, `robotstxt`: robots.txt, `crawldiagnostics`: 404s/redirects)\n",
    "  - only URLs in certain top-level domains\n",
    "\n",
    "  Simply add the restrictions to the WHERE-clause of the join query.\n",
    "\n",
    "- if the URL list isn't really random but does belong to a set of host or domain names: performing the join on the columns `url_host_name` or `url_host_registered_domain` might be more efficient. Cf. [count-domains-alexa-top-1m.sql](https://github.com/commoncrawl/cc-index-table/blob/master/src/sql/examples/cc-index/count-domains-alexa-top-1m.sql) for a join on the domain name.\n",
    "\n",
    "- if the list of URLs or domains is rather short, using [Presto array functions](https://prestodb.io/docs/current/functions/array.html#array-functions) is simpler than a table join:\n",
    "  ```sql\n",
    "  SELECT *\n",
    "  FROM ccindex\n",
    "  WHERE contains(ARRAY ['example.com', '...'], url_host_registered_domain);\n",
    "  ```\n",
    "\n",
    "- if it's about to get all URLs with a certain path prefix, eg., get all blog posts published in 2021 on the Common Crawl site, using a prefix look-up using the [Presto string function](https://prestodb.io/docs/current/functions/string.html) `strpos`:\n",
    "  ```\n",
    "  WHERE strpos(url_surtkey, 'org,commoncrawl)/2021/') = 1\n",
    "  ```\n",
    "  Note that Presto functions can be also used in the `JOIN ON` clause.\n",
    "\n",
    "\n",
    "### Preparing the URL List\n",
    "\n",
    "First, we need to prepare the URL list:\n",
    "1. add the SURT URL as a column\n",
    "2. convert it into a tabular format (Parquet)\n",
    "3. upload the data to S3 and\n",
    "4. import it as table into Athena\n",
    "\n",
    "\n",
    "#### Adding the SURT URL"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "fuzzy-worker",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'com,example)/path/file?a=c&b=d'"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import surt\n",
    "\n",
    "surt.surt('https://www.example.com/path/file?b=d&a=c')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "approved-liberty",
   "metadata": {},
   "source": [
    "#### Writing Parquet\n",
    "\n",
    "If the URL list isn't too big or is partitioned, the simplest way to convert into a Parquet table is to use [pandas](https://pandas.pydata.org/). We first add two columns to our example list holding the SURT URL and the host name:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "spiritual-clearing",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>url_surtkey</th>\n",
       "      <th>url</th>\n",
       "      <th>url_host_name</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>org,commoncrawl)/</td>\n",
       "      <td>http://commoncrawl.org/</td>\n",
       "      <td>commoncrawl.org</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>org,commoncrawl)/2021/02/january-2021-crawl-ar...</td>\n",
       "      <td>https://commoncrawl.org/2021/02/january-2021-c...</td>\n",
       "      <td>commoncrawl.org</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>org,example)/</td>\n",
       "      <td>http://example.org/</td>\n",
       "      <td>example.org</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>org,example)/</td>\n",
       "      <td>http://www.example.org:80</td>\n",
       "      <td>www.example.org:80</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>org,example)/</td>\n",
       "      <td>https://www.example.org/</td>\n",
       "      <td>www.example.org</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>org,example)/path/file?a=c&amp;b=d</td>\n",
       "      <td>https://www.example.org/path/file?b=d&amp;a=c</td>\n",
       "      <td>www.example.org</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                         url_surtkey  \\\n",
       "4                                  org,commoncrawl)/   \n",
       "5  org,commoncrawl)/2021/02/january-2021-crawl-ar...   \n",
       "0                                      org,example)/   \n",
       "3                                      org,example)/   \n",
       "2                                      org,example)/   \n",
       "1                     org,example)/path/file?a=c&b=d   \n",
       "\n",
       "                                                 url       url_host_name  \n",
       "4                            http://commoncrawl.org/     commoncrawl.org  \n",
       "5  https://commoncrawl.org/2021/02/january-2021-c...     commoncrawl.org  \n",
       "0                                http://example.org/         example.org  \n",
       "3                          http://www.example.org:80  www.example.org:80  \n",
       "2                           https://www.example.org/     www.example.org  \n",
       "1          https://www.example.org/path/file?b=d&a=c     www.example.org  "
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from urllib.parse import urlparse\n",
    "\n",
    "import pandas as pd\n",
    "import surt\n",
    "\n",
    "urls = ['http://example.org/',\n",
    "        'https://www.example.org/path/file?b=d&a=c',\n",
    "        'https://www.example.org/',\n",
    "        'http://www.example.org:80',\n",
    "        'http://commoncrawl.org/',\n",
    "        'https://commoncrawl.org/2021/02/january-2021-crawl-archive-now-available/']\n",
    "\n",
    "df = pd.DataFrame(data={'url': urls})\n",
    "\n",
    "df['url_surtkey'] = df['url'].apply(surt.surt)\n",
    "df['url_host_name'] = df['url'].apply(lambda url: urlparse(url).netloc.lower().lstrip('.'))\n",
    "\n",
    "# reorder column and sort by SURT\n",
    "df = df[['url_surtkey', 'url', 'url_host_name']].sort_values(['url_surtkey', 'url'])\n",
    "df"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "continental-notice",
   "metadata": {},
   "source": [
    "and write the table as Parquet:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "hydraulic-appreciation",
   "metadata": {},
   "outputs": [],
   "source": [
    "df.to_parquet('urls.parquet.gz', compression='gzip', index=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "chronic-baltimore",
   "metadata": {},
   "source": [
    "#### Import URL List as Athena Table\n",
    "\n",
    "Now the URL table is uploaded to S3:\n",
    "```\n",
    "aws s3 cp urls.parquet.gz s3://mybucket/myjoin/urls/\n",
    "```\n",
    "Note: `mybucket` is a placeholder - create a bucket on your AWS account in the `us-east-1` region and change the bucket name accordingly.\n",
    "\n",
    "Then we navigate to the [Athena query editor](https://console.aws.amazon.com/athena/home?region=us-east-1#/query-editor) and\n",
    "- create a database \"myjoin\" by executing the following statement:\n",
    "  ```sql\n",
    "  CREATE DATABASE myjoin;\n",
    "  ```\n",
    "- register the table \"urls\":\n",
    "  ```sql\n",
    "  CREATE EXTERNAL TABLE IF NOT EXISTS myjoin.urls (\n",
    "    `url_surtkey`   string,\n",
    "    `url`           string,\n",
    "    `url_host_name` string\n",
    "  )\n",
    "  ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'\n",
    "  WITH SERDEPROPERTIES (\n",
    "    'serialization.format' = '1'\n",
    "  ) LOCATION 's3://mybucket/myjoin/urls/'\n",
    "  TBLPROPERTIES ('has_encrypted_data'='false');\n",
    "  ```\n",
    "- and verify whether the table is imported properly and contains the expected number of rows\n",
    "  ```sql\n",
    "  SELECT * FROM myjoin.urls limit 10;\n",
    "  \n",
    "  SELECT COUNT(*) FROM myjoin.urls;\n",
    "  ```"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "third-circumstances",
   "metadata": {},
   "source": [
    "### Importing the Common Crawl Index Table\n",
    "\n",
    "See [instructions](https://commoncrawl.org/2018/03/index-to-warc-files-and-urls-in-columnar-format/) and the [up-to-date CREATE TABLE statement](https://github.com/commoncrawl/cc-index-table/blob/master/src/sql/athena/cc-index-create-table-flat.sql). If done the index table should be ready under the name \"ccindex.ccindex\".\n",
    "\n",
    "\n",
    "### Joining the Tables\n",
    "\n",
    "Now we are ready to join our URL table with Common Crawl's main table and create a new table with all captures for our URLs:\n",
    "\n",
    "```sql\n",
    "CREATE TABLE myjoin.captures_cc\n",
    "WITH (external_location = 's3://mybucket/myjoin/captures_cc/',\n",
    "      format = 'PARQUET',\n",
    "      parquet_compression = 'GZIP')\n",
    "AS SELECT cc.url_surtkey     AS url_surtkey,\n",
    "       cc.url                AS url,\n",
    "       my.url                AS my_url,\n",
    "       cc.url_host_name      AS url_host_name,\n",
    "       cc.warc_filename      AS warc_filename,\n",
    "       cc.warc_record_offset AS warc_record_offset,\n",
    "       cc.warc_record_length AS warc_record_length,\n",
    "       cc.crawl              AS crawl,\n",
    "       cc.subset             AS subset\n",
    "FROM ccindex.ccindex AS cc\n",
    "  INNER JOIN myjoin.urls AS my\n",
    "  ON my.url_surtkey = cc.url_surtkey\n",
    "WHERE cc.crawl = 'CC-MAIN-2021-10'\n",
    "  AND cc.subset = 'warc'\n",
    "  AND cc.url_host_tld = 'org'\n",
    "```\n",
    "\n",
    "Because all our sample URLs are in the `.org` TLD we can optimize the join which makes it succeed fast - the join query succeeded within 7 seconds scanning only 400 MiB of data.\n",
    "\n",
    "We defined Parquet as output format of the joined table. That's ideal to further inspect the joined data. But Athena supports also other [output formats](https://docs.aws.amazon.com/athena/latest/ug/create-table-as.html#ctas-table-properties) (JSON, AVRO, ORC and other).\n",
    "\n",
    "Let's look what we have in the joined table:\n",
    "```sql \n",
    "SELECT url_surtkey, url, my_url FROM myjoin.captures_cc LIMIT 100;\n",
    "```\n",
    "\n",
    "Not all URLs are found, some URLs have multiple captures and the captured URLs may differ from those in the list:\n",
    "\n",
    "|url_surtkey             |url                     |my_url                       |\n",
    "|:-----------------------|:-----------------------|:----------------------------|\n",
    "|org,commoncrawl)/2021/02/january-2021-crawl-archive-now-available|https://commoncrawl.org/2021/02/january-2021-crawl-archive-now-available/|https://commoncrawl.org/2021/02/january-2021-crawl-archive-now-available/|\n",
    "|org,commoncrawl)/       |http://commoncrawl.org/ |http://commoncrawl.org/      |\n",
    "|org,example)/           |https://example.org/    |https://www.example.org/     |\n",
    "|org,example)/           |https://example.org/    |http://example.org/          |\n",
    "|org,example)/           |https://example.org/    |http://www.example.org:80    |\n",
    "\n",
    "\n",
    "## What's Next? – Extracting the Content from WARC Files\n",
    "\n",
    "The result of the JOIN query include the coordinates of the WARC records holding the captures of the web pages. The three fields `warc_filename`, `warc_record_offset` and `warc_record_length` allow to fetch just the WARC records of interest using range queries.\n",
    "\n",
    "Examples how to do this at scale are included in\n",
    "- [cc-index-table](https://github.com/commoncrawl/cc-index-table#export-subsets-of-the-common-crawl-archives)\n",
    "- [cc-pyspark](https://github.com/commoncrawl/cc-pyspark)\n",
    "\n",
    "Just for demonstration purposes, how to fetch the above mentioned page capture:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "generic-property",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[b'<title>January 2021 crawl archive now available &#8211; Common Crawl</title>']"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import io\n",
    "import re\n",
    "\n",
    "import requests\n",
    "import warcio\n",
    "\n",
    "warc_filename = 'crawl-data/CC-MAIN-2021-10/segments/1614178365186.46/warc/CC-MAIN-20210303012222-20210303042222-00595.warc.gz'\n",
    "warc_record_offset = 250975924\n",
    "warc_record_length = 6922\n",
    "\n",
    "response = requests.get(f'https://data.commoncrawl.org/{warc_filename}',\n",
    "                        headers={'Range': f'bytes={warc_record_offset}-{warc_record_offset + warc_record_length - 1}'})\n",
    "\n",
    "with io.BytesIO(response.content) as stream:\n",
    "    for record in warcio.ArchiveIterator(stream):\n",
    "        html = record.content_stream().read()\n",
    "\n",
    "re.findall(b'<title>.+?</title>', html)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "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.9.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}