# Working with the data

## Using Preprocessing Results stored in Parquet Files 

Preprocessing extracts main text and metadata from WARC content using [resiliparse](https://resiliparse.chatnoir.eu/). The parquet files can then be subsequently used for a number of tasks. Below we provide some simple scripts for working with the data. Note that you could use a S3 file system package like [ffsspec](https://pypi.org/project/fsspec/) to work with minio buckets directly.

```python 
import os
import pandas as pd
import glob
import logging

# Setup logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# Define the directory path
dir_path = "./parquet/"

# Getting all the parquet files from the directory
files = glob.glob(os.path.join(dir_path, "*.parquet"))

# Initialize counters and frequency distributions
total_unique_url = 0
total_unique_content_type = set()
total_unique_http_server = set()
total_unique_language = set()

# Initialize frequency dataframes
df_content_type_freq = None
df_http_server_freq = None
df_language_freq = None
df_url_freq = None
df_tld_freq = None

# 1. Read parquet files one by one, generate statistics, and aggregate them
for file in files:
    logging.info(f"Processing file: {file}")

    # Read parquet file
    data = pd.read_parquet(file)

    # Update url combination statistics
    unique_urls = data['url_subdomain'] + "." +data['url_domain'] + "." + data['url_suffix']
    data["unique_urls"] = data['url_subdomain'] + "." +data['url_domain'] + "." + data['url_suffix']
    total_unique_url += unique_urls.nunique()

    # Update 'http_content_type', 'http_server', and 'language' statistics
    total_unique_content_type.update(data['http_content_type'].unique())
    total_unique_http_server.update(data['http_server'].unique())
    total_unique_language.update(data['language'].unique())

    # Update frequency distributions
    df_tld_freq = df_tld_freq.add(data['url_suffix'].value_counts(), fill_value=0) if df_tld_freq is not None else data['unique_urls'].value_counts()
    df_url_freq = df_url_freq.add(data['unique_urls'].value_counts(), fill_value=0) if df_url_freq is not None else data['unique_urls'].value_counts()
    df_content_type_freq = df_content_type_freq.add(data['http_content_type'].value_counts(), fill_value=0) if df_content_type_freq is not None else data['http_content_type'].value_counts()
    df_http_server_freq = df_http_server_freq.add(data['http_server'].value_counts(), fill_value=0) if df_http_server_freq is not None else data['http_server'].value_counts()
    df_language_freq = df_language_freq.add(data['language'].value_counts(), fill_value=0) if df_language_freq is not None else data['language'].value_counts()

# Convert set lengths to DataFrame for easy writing to TSV
df_url_counts = pd.DataFrame({"unique_url_combination_counts": [total_unique_url]})
df_content_type_counts = pd.DataFrame({"unique_http_content_type_counts": [len(total_unique_content_type)]})
df_http_server_counts = pd.DataFrame({"unique_http_server_counts": [len(total_unique_http_server)]})
df_language_counts = pd.DataFrame({"unique_language_counts": [len(total_unique_language)]})

# Transpose frequency dataframes
df_content_type_freq = df_content_type_freq.transpose()
df_http_server_freq = df_http_server_freq.transpose()
df_language_freq = df_language_freq.transpose()

# 3. Write one TSV file per statistic into the same directory with a proper name
df_url_counts.to_csv(os.path.join(dir_path, "unique_url_counts.tsv"), sep="\t", index=False)
df_content_type_counts.to_csv(os.path.join(dir_path, "unique_content_type_counts.tsv"), sep="\t", index=False)
df_http_server_counts.to_csv(os.path.join(dir_path, "unique_http_server_counts.tsv"), sep="\t", index=False)
df_language_counts.to_csv(os.path.join(dir_path, "unique_language_counts.tsv"), sep="\t", index=False)

# Save frequency distributions to TSV files
df_url_freq.to_csv(os.path.join(dir_path, "url_frequency.tsv"), sep="\t")
df_tld_freq.to_csv(os.path.join(dir_path, "tld_frequency.tsv"), sep="\t")
df_content_type_freq.to_csv(os.path.join(dir_path, "content_type_frequency.tsv"), sep="\t")
df_http_server_freq.to_csv(os.path.join(dir_path, "http_server_frequency.tsv"), sep="\t")
df_language_freq.to_csv(os.path.join(dir_path, "language_frequency.tsv"), sep="\t")

logging.info("Processing completed. Results are saved as TSV files.")
```

Another option for working with parquet files would be to invoke different S3 Select commands. Note that minio does not implement a query engine, but usually iterates row wise through the parquet files and apply the specified query filter. So [S3 Select commands](https://blog.min.io/tag/s3-select/) can use significant resources. 

For example, when using the minio-client (if enabled by the server):

```bash
mc sql --recursive --query "select count(*) from S3Object" <alias>/path/preprocessed/2023-10
```