CORDIS2parquet.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "14dfa89a",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# from pyspark.sql import SparkSession\n",
    "\n",
    "# spark = SparkSession.builder.appName(\"test\").getOrCreate()\n",
    "# sc = spark.sparkContext\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "79efdd41",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "sc.version\n",
    "spark.conf.set(\"spark.sql.parquet.int96RebaseModeInWrite\", \"CORRECTED\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6808f216",
   "metadata": {},
   "source": [
    "# Imports"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "de1121c4",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import pandas as pd\n",
    "from configparser import ConfigParser\n",
    "from pathlib import Path\n",
    "import zipfile\n",
    "# import py7zr\n",
    "import io\n",
    "import os\n",
    "import re\n",
    "from pyspark.sql.functions import col\n",
    "from pyspark.sql import functions as F\n",
    "from pyspark.sql.types import StringType\n",
    "from typing import List\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "827c6989",
   "metadata": {},
   "source": [
    "# Auxiliary Functions\n",
    "\n",
    "These functions are used to load, read, identify or save files."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0bb901a7",
   "metadata": {},
   "source": [
    "## File processing"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "0c18776c",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def get_only_letter(s):\n",
    "    res = re.sub(r\"[^a-zA-Z]\", \"\", s)\n",
    "    return res\n",
    "\n",
    "\n",
    "def get_file_extension(f: Path):\n",
    "    output = os.popen(f\"file -b {f.as_posix()}\")\n",
    "    return output.read().split()[0]\n",
    "\n",
    "\n",
    "def read_excel(f: Path):\n",
    "    fname = None\n",
    "    df = None\n",
    "    try:\n",
    "        fname = f.stem\n",
    "        df = pd.read_excel(f, index_col=False)\n",
    "        df.columns = df.columns.str.strip()\n",
    "\n",
    "    except Exception as e:\n",
    "        print(\"-- ERROR: file could not be processed\")\n",
    "        print(\"\\t\", e)\n",
    "    return fname, df\n",
    "\n",
    "\n",
    "def extract_file_info(f: Path):\n",
    "    \"\"\"\n",
    "    Read a file and extract all its content into a pd.DataFrame()\n",
    "\n",
    "    As a zip file can contain multiple inner files, we need to iterate\n",
    "    over this function when calling it\n",
    "    \"\"\"\n",
    "    fname = None\n",
    "    df = None\n",
    "\n",
    "    # Get type of file\n",
    "    extension = get_file_extension(f)\n",
    "    print(f\"-- {extension}\")\n",
    "\n",
    "    if extension not in [\"Zip\", \"7-zip\"]:\n",
    "        # Excel files (hopefully)\n",
    "        fname, df = read_excel(f)\n",
    "        yield fname, df\n",
    "    else:\n",
    "        # Compressed files\n",
    "        if extension == \"Zip\":\n",
    "            archive = zipfile.ZipFile(f.as_posix(), \"r\")\n",
    "            content = {name: archive.read(name) for name in archive.namelist()}\n",
    "        elif extension == \"7-zip\":\n",
    "            with py7zr.SevenZipFile(f, \"r\") as archive:\n",
    "                content = archive.readall()\n",
    "        print(\"-- Files:\")\n",
    "        keys = [get_only_letter(k) for k in content.keys()]\n",
    "        key_0 = keys[0]\n",
    "\n",
    "        # Check if all elements are of the same table:\n",
    "        all_equal = True if all(key_0 == x for x in keys) else False\n",
    "\n",
    "        # Return info\n",
    "        df_sub = []\n",
    "        for k, v in content.items():\n",
    "            print(f\"-- -- {k}\")\n",
    "            df = pd.read_excel(v, engine=\"openpyxl\")\n",
    "            if all_equal:\n",
    "                df_sub.append(df)\n",
    "                continue\n",
    "            else:\n",
    "                fname = f.stem + \"-\" + re.sub(\"xlsx\", \"\", get_only_letter(k))\n",
    "                yield fname, df\n",
    "\n",
    "        # Merge all into one dataframe if necessary\n",
    "        if all_equal:\n",
    "            fname = key_0\n",
    "            df = pd.concat(df_sub, ignore_index=True)\n",
    "            yield fname, df\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c76001b0",
   "metadata": {},
   "source": [
    "## Spark conversion"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "4ed218e4",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def convert_spark(df: pd.DataFrame):\n",
    "    \"\"\"\n",
    "    Transforms input pandas DataFrame into spark DataFrame\n",
    "    \"\"\"\n",
    "    sparkDF = None\n",
    "    \n",
    "    # Clean DF\n",
    "    # Strip column names\n",
    "    df.columns = df.columns.str.strip()\n",
    "\n",
    "    # Filter languages\n",
    "    if \"language\" in df.columns.str.lower():\n",
    "        df = df[(df.loc[:, df.columns.str.lower() == \"language\"] == \"en\").iloc[:, 0]]\n",
    "\n",
    "    # Remove empty columns\n",
    "    df = df.drop([c for c in df.columns if c.startswith(\"Unnamed:\")], axis=1)\n",
    "\n",
    "    # Replace NaN\n",
    "    df = df.replace({np.nan: None})\n",
    "    df.loc[:, df.isnull().all()] = \"\"\n",
    "\n",
    "    # Format dates\n",
    "    # date_cols = [\"date\" in i for i in df.columns.str.lower()]\n",
    "    # df.loc[:, date_cols] = df.loc[:, date_cols].apply(pd.to_datetime, errors=\"coerce\")\n",
    "    date_cols = [c for c in df.columns if \"date\" in c.lower()]\n",
    "    df[date_cols] = df[date_cols].apply(pd.to_datetime, errors=\"coerce\")\n",
    "\n",
    "    # Convert\n",
    "    try:\n",
    "        sparkDF = spark.createDataFrame(df)\n",
    "        sparkDF = sparkDF.replace(\"\", None)\n",
    "        print(\"-- Conversion OK\")\n",
    "\n",
    "    except (TypeError, ValueError):\n",
    "        print(\"-- -- Transform\")\n",
    "        df = pd.read_csv(io.StringIO(df.to_csv(index=False)))\n",
    "        # Replace NaN\n",
    "        df = df.replace({np.nan: None})\n",
    "        df.loc[:, df.isnull().all()] = \"\"\n",
    "\n",
    "        sparkDF = spark.createDataFrame(df)\n",
    "        sparkDF = sparkDF.replace(\"\", None)\n",
    "        print(\"-- Conversion OK\")\n",
    "\n",
    "    except Exception as e:\n",
    "        print(\"-- Exception\")\n",
    "        print(\"\\t\", e.__class__.__name__, \"\\n\\t\", e)\n",
    "\n",
    "    return sparkDF\n",
    "\n",
    "\n",
    "def save_parquet(sparkDF, fname):\n",
    "    # Save original dataframe\n",
    "    print(f'-- Saving in {dir_parquet.joinpath(f\"{fname}.parquet\")}')\n",
    "    sparkDF.write.parquet(\n",
    "        dir_parquet.joinpath(f\"{fname}.parquet\").as_posix(),\n",
    "        mode=\"overwrite\",\n",
    "    )\n",
    "    print(\"SAVE SUCCESS\")\n",
    "\n",
    "\n",
    "def load_parquet(fname):\n",
    "    # Load dataframe\n",
    "    print(f'-- Loading {dir_parquet.joinpath(f\"{fname}.parquet\")}')\n",
    "    sparkDF = spark.read.parquet(dir_parquet.joinpath(f\"{fname}.parquet\").as_posix())\n",
    "    return sparkDF\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "eb0ca8fc",
   "metadata": {},
   "source": [
    "# Define directories\n",
    "\n",
    "In the script version the directory is passed through configuration file. \n",
    "\n",
    "Here, you need to write directly the location of:\n",
    "\n",
    "  - dir_raw     : the directory containing the raw data\n",
    "  - dir_parquet : the directory in HDFS that will contain the processed files\n",
    "\n",
    "If join with Semantic Scholar and PATSTAT is activated, you need to specify also the location of these files in the HDFS"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "917cb454-6dee-4982-8ac4-2414da23d9b9",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Define directories\n",
    "\n",
    "# cf = ConfigParser()\n",
    "# cf.read(\"config.cf\")\n",
    "\n",
    "# Data sources\n",
    "# dir_raw = Path(\"/export/data_ml4ds/IntelComp/Datasets/cordis/20230425 \")\n",
    "dir_raw = Path(\"/export/data_ml4ds/IntelComp/Datasets/cordis/20230823/rawdata\")\n",
    "\n",
    "# Auxiliary datasets\n",
    "ss_join = True\n",
    "# dir_ss = Path(cf.get(\"aux\", \"dir_ss\"))\n",
    "dir_ss = Path(\"/export/ml4ds/IntelComp/Datalake/semanticscholar/20230418/parquet\")\n",
    "pt_join = True\n",
    "# dir_patstat = Path(cf.get(\"aux\", \"dir_patstat\"))\n",
    "dir_patstat = Path(\"/export/ml4ds/IntelComp/Datalake/patstat/2023_Spring/parquet\")\n",
    "\n",
    "# Target directory\n",
    "# dir_parquet = Path(cf.get(\"cordis\", \"dir_parquet\"))\n",
    "dir_parquet = Path(\"/export/ml4ds/IntelComp/Datalake/CORDIS/20230823/parquet\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "e17a8622",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "dir_raw_horizon = dir_raw.joinpath(\"HORIZON\")\n",
    "dir_raw_h2020 = dir_raw.joinpath(\"H2020\")\n",
    "dir_raw_fp7 = dir_raw.joinpath(\"FP7\")\n",
    "dir_raw_ref = dir_raw.joinpath(\"cordis-ref\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "d6d909e1",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Configuration hdfs\n",
    "fs = spark._jvm.org.apache.hadoop.fs.FileSystem.get(spark._jsc.hadoopConfiguration())\n",
    "hdfs_dir_parquet = spark._jvm.org.apache.hadoop.fs.Path(dir_parquet.as_posix())\n",
    "# hdfs_dir_raw = spark._jvm.org.apache.hadoop.fs.Path(dir_raw.as_posix())\n",
    "\n",
    "# Create output directories if they do not exist\n",
    "if not fs.exists(hdfs_dir_parquet):\n",
    "    fs.mkdirs(hdfs_dir_parquet)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b3220544",
   "metadata": {},
   "source": [
    "# Process Files"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "57575a71",
   "metadata": {},
   "source": [
    "Save all tables to parquet"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6d2629d2",
   "metadata": {},
   "source": [
    "## Read SemanticScholar\n",
    "\n",
    "Load SemanticScholar information. It will be joint with Publications"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "637982d1",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# ss = spark.read.parquet(\"data/sample.parquet\")\n",
    "if ss_join:\n",
    "    ss = spark.read.parquet(dir_ss.joinpath(\"papers.parquet\").as_posix())\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dac4a8ac",
   "metadata": {},
   "source": [
    "## Read PATSTAT\n",
    "\n",
    "Load PATSTAT information. It will be joint with Patents"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "96cab171",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "if pt_join:\n",
    "    pt = spark.read.parquet(dir_patstat.joinpath(\"patstat_appln.parquet\").as_posix())\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "10e029bc",
   "metadata": {},
   "source": [
    "## File list"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "54ff6952",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Files in LOCAL:\n",
    "# Read fp7 files\n",
    "fp7_file_list = [el for el in dir_raw_fp7.iterdir()]\n",
    "\n",
    "# Read h2020 files\n",
    "h2020_file_list = [el for el in dir_raw_h2020.iterdir()]\n",
    "\n",
    "# Read horizon files\n",
    "horizon_file_list = [el for el in dir_raw_horizon.iterdir()]\n",
    "\n",
    "file_list = [\n",
    "    el\n",
    "    for el in (fp7_file_list + h2020_file_list + horizon_file_list)\n",
    "    if (\n",
    "        el.name.endswith(\"xlsx.zip\")\n",
    "        or el.name.endswith(\"xlsx.7z\")\n",
    "        or el.name.endswith(\".xls\")\n",
    "        or el.name.endswith(\".xlsx\")\n",
    "    )\n",
    "]\n",
    "\n",
    "publications = [x for x in file_list if \"publications\" in x.name.lower()]\n",
    "reports = [x for x in file_list if \"reports\" in x.name.lower()]\n",
    "projects = [x for x in file_list if \"projects\" in x.name.lower()]\n",
    "irps = [x for x in file_list if \"irps\" in x.name.lower()]\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "00eb0efa-6438-433a-9bc6-54155fee7359",
   "metadata": {},
   "source": [
    "## Processing Functions"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2dd8e804",
   "metadata": {},
   "source": [
    "### Publications"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "33beaa00",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Processing...\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- Microsoft\n",
      "-- FP7PC_DM_PROJ_PUBLICATIONS\n",
      "\n",
      " -------------------------------------------------------------------------------- \n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- Zip\n",
      "-- Files:\n",
      "-- -- xlsx/projectPublications.xlsx\n",
      "-- -- xlsx/projectPublications_2.xlsx\n",
      "-- -- xlsx/projectPublications_3.xlsx\n",
      "-- -- xlsx/projectPublications_4.xlsx\n",
      "-- -- xlsx/projectPublications_5.xlsx\n",
      "-- -- xlsx/projectPublications_6.xlsx\n",
      "-- -- xlsx/projectPublications_7.xlsx\n",
      "-- xlsxprojectPublicationsxlsx\n",
      "\n",
      " -------------------------------------------------------------------------------- \n",
      "\n",
      "-- Conversion OK\n",
      "-- Saving in /export/ml4ds/IntelComp/Datalake/CORDIS/20230823/parquet/publications.parquet\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "23/08/26 21:27:30 WARN TaskSetManager: Stage 3 contains a task of very large size (4820 KiB). The maximum recommended task size is 1000 KiB.\n",
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SAVE SUCCESS\n"
     ]
    }
   ],
   "source": [
    "def cleanDOI(doi):\n",
    "    if not isinstance(doi, str):\n",
    "        doi = f\"{doi}\".lower()\n",
    "    doi = re.sub(r\"^https://doi.org/\", \"\", doi.lower())\n",
    "    doi = \"\".join(doi.split())\n",
    "    return doi\n",
    "\n",
    "\n",
    "cleanDOI_udf = F.udf(cleanDOI, StringType())\n",
    "\n",
    "\n",
    "def process_publications(publications: List[Path], merge=False, ss_join=False):\n",
    "    \"\"\"\n",
    "    Processes a list of Path, where each element is a file path with publications.\\\\\n",
    "    (Optional) Concatenate all publications in one dataframe and join with SemanticScholar.\\\\\n",
    "    Then, save it to parquet.\n",
    "\n",
    "    If `merge`=`True` merge all dataframes into one\n",
    "    \n",
    "    If `ss_join`=`True` includes the IDs of Semantic Scholar publications in dataframe\n",
    "    \"\"\"\n",
    "\n",
    "    pubs = []\n",
    "\n",
    "    print(\"Processing...\\n\")\n",
    "    for f in publications:\n",
    "        for fname, df in extract_file_info(f):\n",
    "            print(f\"-- {fname}\")\n",
    "\n",
    "            if fname is None or df is None:\n",
    "                continue\n",
    "\n",
    "            # Unify format\n",
    "            df.columns = df.columns.str.lower()\n",
    "            if \"horizon\" in f.name.lower():\n",
    "                df[\"frameworkProgramme\"] = \"HORIZON\"\n",
    "                df = df.rename(\n",
    "                    columns={\n",
    "                        \"id\": \"id\",\n",
    "                        \"title\": \"title\",\n",
    "                        \"ispublishedas\": \"isPublishedAs\",\n",
    "                        \"authors\": \"authors\",\n",
    "                        \"journaltitle\": \"journalTitle\",\n",
    "                        \"journalnumber\": \"journalNumber\",\n",
    "                        \"publishedyear\": \"publishedYear\",\n",
    "                        \"publishedpages\": \"publishedPages\",\n",
    "                        \"issn\": \"issn\",\n",
    "                        \"isbn\": \"isbn\",\n",
    "                        \"doi\": \"doi\",\n",
    "                        \"projectid\": \"projectID\",\n",
    "                        \"projectacronym\": \"projectAcronym\",\n",
    "                        \"collection\": \"collection\",\n",
    "                        \"contentupdatedate\": \"contentUpdateDate\",\n",
    "                        \"rcn\": \"rcn\",\n",
    "                    }\n",
    "                )\n",
    "            elif \"h2020\" in f.name.lower():\n",
    "                df[\"frameworkProgramme\"] = \"H2020\"\n",
    "                df = df.rename(\n",
    "                    columns={\n",
    "                        \"id\": \"id\",\n",
    "                        \"title\": \"title\",\n",
    "                        \"ispublishedas\": \"isPublishedAs\",\n",
    "                        \"authors\": \"authors\",\n",
    "                        \"journaltitle\": \"journalTitle\",\n",
    "                        \"journalnumber\": \"journalNumber\",\n",
    "                        \"publishedyear\": \"publishedYear\",\n",
    "                        \"publishedpages\": \"publishedPages\",\n",
    "                        \"issn\": \"issn\",\n",
    "                        \"isbn\": \"isbn\",\n",
    "                        \"doi\": \"doi\",\n",
    "                        \"projectid\": \"projectID\",\n",
    "                        \"projectacronym\": \"projectAcronym\",\n",
    "                        \"collection\": \"collection\",\n",
    "                        \"contentupdatedate\": \"contentUpdateDate\",\n",
    "                        \"rcn\": \"rcn\",\n",
    "                    }\n",
    "                )\n",
    "            else:\n",
    "                df[\"frameworkProgramme\"] = \"FP7\"\n",
    "                df = df.drop(columns=\"doi\").rename(columns={\"qa_processed_doi\": \"doi\"})\n",
    "                df = df.rename(\n",
    "                    columns={\n",
    "                        \"project_id\": \"projectID\",\n",
    "                        \"title\": \"title\",\n",
    "                        \"author\": \"authors\",\n",
    "                        # \"doi\": \"\",\n",
    "                        \"publication_type\": \"isPublishedAs\",\n",
    "                        \"repository_url\": \"repositoryUrl\",\n",
    "                        \"journal_title\": \"journalTitle\",\n",
    "                        \"publisher\": \"publisher\",\n",
    "                        \"volume\": \"journalNumber\",\n",
    "                        \"pages\": \"publishedPages\",\n",
    "                        \"qa_processed_doi\": \"doi\",\n",
    "                        \"record_id\": \"id\",\n",
    "                    }\n",
    "                )\n",
    "            if merge:\n",
    "                pubs.append(df)\n",
    "            else:\n",
    "                df[\"doi\"] = df[\"doi\"].apply(cleanDOI)\n",
    "                # Transform to spark\n",
    "                sparkDF = convert_spark(df)\n",
    "\n",
    "                # Join with SemanticScholar\n",
    "                if ss_join:\n",
    "                    joint_pub = (\n",
    "                        sparkDF.withColumn(\n",
    "                            \"doi\",\n",
    "                            F.when(col(\"doi\").isNotNull(), cleanDOI_udf(col(\"doi\"))).otherwise(None),\n",
    "                        )\n",
    "                        .join(\n",
    "                            ss.select(\n",
    "                                col(\"id\").alias(\"SSID\"),\n",
    "                                F.when(col(\"doi\").isNotNull(), cleanDOI_udf(col(\"doi\")))\n",
    "                                .otherwise(None)\n",
    "                                .alias(\"doi\"),\n",
    "                            ),\n",
    "                            on=\"doi\",\n",
    "                            how=\"left\",\n",
    "                        )\n",
    "                        .select(df.columns.tolist() + [\"SSID\"])\n",
    "                    )\n",
    "                else:\n",
    "                    joint_pub = sparkDF\n",
    "\n",
    "                save_parquet(joint_pub, fname)\n",
    "\n",
    "        print(\"\\n\", \"-\" * 80, \"\\n\")\n",
    "\n",
    "    if merge:\n",
    "        pub_df = pd.concat(pubs)\n",
    "        \n",
    "        # Transform to spark\n",
    "        sparkDF = convert_spark(pub_df)\n",
    "        \n",
    "        # Join with SemanticScholar\n",
    "        if ss_join:\n",
    "            joint_pub = (\n",
    "                sparkDF.withColumn(\n",
    "                    \"doi\",\n",
    "                    F.when(col(\"doi\").isNotNull(), cleanDOI_udf(col(\"doi\"))).otherwise(None),\n",
    "                )\n",
    "                .join(\n",
    "                    ss.select(\n",
    "                        col(\"id\").alias(\"SSID\"),\n",
    "                        F.when(col(\"doi\").isNotNull(), cleanDOI_udf(col(\"doi\")))\n",
    "                        .otherwise(None)\n",
    "                        .alias(\"doi\"),\n",
    "                    ),\n",
    "                    on=\"doi\",\n",
    "                    how=\"left\",\n",
    "                )\n",
    "                .select(pub_df.columns.tolist() + [\"SSID\"])\n",
    "            )\n",
    "        else:\n",
    "            joint_pub = sparkDF\n",
    "\n",
    "        \n",
    "        save_parquet(joint_pub, \"publications\")\n",
    "\n",
    "\n",
    "process_publications(publications, merge=True, ss_join=ss_join)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8f46b844",
   "metadata": {},
   "source": [
    "Some publications statistics:\n",
    "- Percentage of projects with publications\n",
    "- Number of publications with DOI\n",
    "- Number of publications with reference in SemanticScholar"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "a0c5671e",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- Loading /export/ml4ds/IntelComp/Datalake/CORDIS/20230823/parquet/publications.parquet\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Number of total publications: 731356\n",
      "Number of publications with DOI: 626316 (85.64%)\n",
      "\t-- FP7: 320021 (43.76%)\n",
      "\t-- H2020: 306295 (41.88%)\n",
      "Number of publications with SSID: 591208 (80.84%)\n",
      "Number of publications with DOI that don't have SS reference: 35108\n"
     ]
    }
   ],
   "source": [
    "def publications_statistics(ss_join=False):\n",
    "    pubs = load_parquet(\"publications\")\n",
    "    if ss_join:\n",
    "        df = pubs.select(\"frameworkProgramme\", \"doi\", \"SSID\").toPandas()\n",
    "    else:\n",
    "        df = pubs.select(\"frameworkProgramme\", \"doi\").toPandas()\n",
    "                \n",
    "    total = len(df)\n",
    "    print(f\"Number of total publications: {total}\")\n",
    "\n",
    "    # Publications with DOI\n",
    "    pub_doi = df[[\"frameworkProgramme\", \"doi\"]].dropna(subset=[\"doi\"])\n",
    "    num_pubs = pub_doi[\"doi\"].count()\n",
    "    print(f\"Number of publications with DOI: {num_pubs} ({num_pubs/total*100:.2f}%)\")\n",
    "    for k, v in pub_doi.groupby(\"frameworkProgramme\")[\"doi\"].count().items():\n",
    "        print(f\"\\t-- {k}: {v} ({v/total*100:.2f}%)\")\n",
    "\n",
    "    # Publications with ref in SS\n",
    "    if ss_join:\n",
    "        pub_SS = df[[\"doi\", \"SSID\"]].dropna(subset=[\"SSID\"])\n",
    "        pub_withSS = len(pub_SS)\n",
    "        print(f\"Number of publications with SSID: {pub_withSS} ({pub_withSS/total*100:.2f}%)\")\n",
    "        print(f\"Number of publications with DOI that don't have SS reference: {num_pubs-pub_withSS}\")\n",
    "\n",
    "\n",
    "publications_statistics(ss_join=ss_join)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "67a24d1d-fe38-47a6-893b-cd9cb4294459",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- Loading /export/ml4ds/IntelComp/Datalake/CORDIS/20220823/parquet/publications.parquet\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# Save XLS file to NFS location\n",
    "# publication_file = \"/export/data_ml4ds/IntelComp/Datasets/cordis/20220823/xlsx/publications.xlsx\"\n",
    "# load_parquet(\"publications\").toPandas().to_excel(publication_file, index=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e55e1a09",
   "metadata": {},
   "source": [
    "### Patents"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "f2858d94",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Processing...\n",
      "\n",
      "-- Microsoft\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- projectIrps_h2020\n",
      "\n",
      " -------------------------------------------------------------------------------- \n",
      "\n",
      "-- Microsoft\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- projectIrps_fp7\n",
      "\n",
      " -------------------------------------------------------------------------------- \n",
      "\n",
      "-- Conversion OK\n",
      "-- Saving in /export/ml4ds/IntelComp/Datalake/CORDIS/20230823/parquet/patents.parquet\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SAVE SUCCESS\n"
     ]
    }
   ],
   "source": [
    "def process_patents(patents: List[Path], merge=False, pt_join=False):\n",
    "    \"\"\"\n",
    "    Processes a list of Path, where each element is a file path with patents.\\\\\n",
    "    (Optional) Concatenate all patents in one dataframe and join with PATSTAT.\\\\\n",
    "    Then, save it to parquet.\n",
    "\n",
    "    If `merge`=`True` merge all dataframes into one\n",
    "    \n",
    "    If `pt_join`=`True` includes the IDs of PATSTAT in dataframe\n",
    "    \"\"\"\n",
    "\n",
    "    pubs = []\n",
    "\n",
    "    print(\"Processing...\\n\")\n",
    "    for f in patents:\n",
    "        for fname, df in extract_file_info(f):\n",
    "            print(f\"-- {fname}\")\n",
    "\n",
    "            if fname is None or df is None:\n",
    "                continue\n",
    "\n",
    "            if merge:\n",
    "                if \"horizon\" in f.name.lower():\n",
    "                    df[\"frameworkProgramme\"] = \"HORIZON\"\n",
    "                elif \"h2020\" in f.name.lower():\n",
    "                    df[\"frameworkProgramme\"] = \"H2020\"\n",
    "                elif \"fp7\" in f.name.lower():\n",
    "                    df[\"frameworkProgramme\"] = \"FP7\"\n",
    "                pubs.append(df)\n",
    "            else:\n",
    "                # Transform to spark\n",
    "                sparkDF = convert_spark(df)\n",
    "\n",
    "                # Join with PATSTAT\n",
    "                if pt_join:\n",
    "                    joint_pat = sparkDF.join(\n",
    "                        pt.select(col(\"appln_nr\").alias(\"applicationIdentifier\"), \"appln_id\"),\n",
    "                        on=\"applicationIdentifier\",\n",
    "                        how=\"left\",\n",
    "                    ).select(df.columns.tolist() + [\"appln_id\"])\n",
    "                else:\n",
    "                    joint_pat = sparkDF\n",
    "\n",
    "                save_parquet(joint_pat, fname)\n",
    "\n",
    "        print(\"\\n\", \"-\" * 80, \"\\n\")\n",
    "\n",
    "    if merge:\n",
    "        pat_df = pd.concat(pubs)\n",
    "        # Transform to spark\n",
    "        sparkDF = convert_spark(pat_df)\n",
    "\n",
    "        # Join with PATSTAT\n",
    "        if pt_join:\n",
    "            joint_pat = sparkDF.join(\n",
    "                pt.select(col(\"appln_nr\").alias(\"applicationIdentifier\"), \"appln_id\"),\n",
    "                on=\"applicationIdentifier\",\n",
    "                how=\"left\",\n",
    "            ).select(pat_df.columns.tolist() + [\"appln_id\"])\n",
    "        else:\n",
    "            joint_pat = sparkDF\n",
    "\n",
    "        save_parquet(joint_pat, \"patents\")\n",
    "\n",
    "\n",
    "process_patents(irps, merge=True, pt_join=pt_join)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "17ce5945",
   "metadata": {},
   "source": [
    "Some patents statistics:\n",
    "- Number of patents with reference in PATSTAT"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "c1545418",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- Loading /export/ml4ds/IntelComp/Datalake/CORDIS/20230823/parquet/patents.parquet\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Number of total patents: 32073\n",
      "\t-- FP7: 23368 (72.86%)\n",
      "\t-- H2020: 8705 (27.14%)\n",
      "Number of patents in PATSTAT: 32064 (99.97%)\n",
      "\t-- FP7: 23368 (72.86%)\n",
      "\t-- H2020: 8696 (27.11%)\n"
     ]
    }
   ],
   "source": [
    "def patents_statistics(pt_join=False):\n",
    "    pats = load_parquet(\"patents\")\n",
    "    if pt_join:\n",
    "        df = pats.select(\"frameworkProgramme\", \"appln_id\").toPandas()\n",
    "    else:\n",
    "        df = pats.select(\"frameworkProgramme\").toPandas()\n",
    "\n",
    "    total = len(df)\n",
    "    print(f\"Number of total patents: {total}\")\n",
    "\n",
    "    for k, v in df.groupby(\"frameworkProgramme\")[\"frameworkProgramme\"].count().items():\n",
    "        print(f\"\\t-- {k}: {v} ({v/total*100:.2f}%)\")\n",
    "\n",
    "    # Patents in PATSTAT\n",
    "    if pt_join:\n",
    "        pat_PATS = df[[\"frameworkProgramme\", \"appln_id\"]].dropna(subset=[\"appln_id\"])\n",
    "        num_pats = pat_PATS[\"appln_id\"].count()\n",
    "        print(f\"Number of patents in PATSTAT: {num_pats} ({num_pats/total*100:.2f}%)\")\n",
    "        for k, v in pat_PATS.groupby(\"frameworkProgramme\")[\"appln_id\"].count().items():\n",
    "            print(f\"\\t-- {k}: {v} ({v/total*100:.2f}%)\")\n",
    "\n",
    "\n",
    "patents_statistics(pt_join=pt_join)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "11052bee-28d6-451d-9301-675c54bc0e54",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Count unique patents according to id\n",
    "# load_parquet(\"patents\").select(\"appln_id\").distinct().count()\n",
    "\n",
    "# Save XLS file to NFS location\n",
    "# patent_file = \"/export/data_ml4ds/IntelComp/Datasets/cordis/20220908/xlsx/patents.xlsx\"\n",
    "# load_parquet(\"patents\").toPandas().to_excel(patent_file, index=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c87ff072",
   "metadata": {},
   "source": [
    "### Summaries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "f51e7ed2",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Processing...\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- Zip\n",
      "-- Files:\n",
      "-- -- xlsx/reportSummaries.xlsx\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/lib/python3.10/site-packages/openpyxl/styles/stylesheet.py:226: UserWarning: Workbook contains no default style, apply openpyxl's default\n",
      "  warn(\"Workbook contains no default style, apply openpyxl's default\")\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- xlsxreportSummariesxlsx\n",
      "\n",
      " -------------------------------------------------------------------------------- \n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- Zip\n",
      "-- Files:\n",
      "-- -- xlsx/reportSummaries.xlsx\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/lib/python3.10/site-packages/openpyxl/styles/stylesheet.py:226: UserWarning: Workbook contains no default style, apply openpyxl's default\n",
      "  warn(\"Workbook contains no default style, apply openpyxl's default\")\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- xlsxreportSummariesxlsx\n",
      "\n",
      " -------------------------------------------------------------------------------- \n",
      "\n",
      "-- Zip\n",
      "-- Files:\n",
      "-- -- xlsx/reportSummaries.xlsx\n",
      "-- xlsxreportSummariesxlsx\n",
      "\n",
      " -------------------------------------------------------------------------------- \n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "/usr/lib/python3.10/site-packages/openpyxl/styles/stylesheet.py:226: UserWarning: Workbook contains no default style, apply openpyxl's default\n",
      "  warn(\"Workbook contains no default style, apply openpyxl's default\")\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- Conversion OK\n",
      "-- Saving in /export/ml4ds/IntelComp/Datalake/CORDIS/20230823/parquet/reports.parquet\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SAVE SUCCESS\n"
     ]
    }
   ],
   "source": [
    "def process_reports(reports: List[Path], merge=False):\n",
    "    \"\"\"\n",
    "    Processes a list of Path, where each element is a file path with report summaries.\\\\\n",
    "    (Optional) Concatenate all patents in one dataframe.\\\\\n",
    "    Then, save it to parquet.\n",
    "\n",
    "\n",
    "    If `merge`=`True` merge all dataframes into one\n",
    "    \"\"\"\n",
    "\n",
    "    reps = []\n",
    "\n",
    "    print(\"Processing...\\n\")\n",
    "    for f in reports:\n",
    "        for fname, df in extract_file_info(f):\n",
    "            print(f\"-- {fname}\")\n",
    "\n",
    "            if fname is None or df is None:\n",
    "                continue\n",
    "\n",
    "            if merge:\n",
    "                if \"horizon\" in f.name.lower():\n",
    "                    df[\"frameworkProgramme\"] = \"HORIZON\"\n",
    "                elif \"h2020\" in f.name.lower():\n",
    "                    df[\"frameworkProgramme\"] = \"H2020\"\n",
    "                elif \"fp7\" in f.name.lower():\n",
    "                    df[\"frameworkProgramme\"] = \"FP7\"\n",
    "                reps.append(df)\n",
    "            else:\n",
    "                # Transform to spark\n",
    "                sparkDF = convert_spark(df)\n",
    "\n",
    "                save_parquet(sparkDF, fname)\n",
    "\n",
    "        print(\"\\n\", \"-\" * 80, \"\\n\")\n",
    "\n",
    "    if merge:\n",
    "        reps_df = pd.concat(reps)\n",
    "        # Transform to spark\n",
    "        sparkDF = convert_spark(reps_df)\n",
    "\n",
    "        save_parquet(sparkDF, \"reports\")\n",
    "\n",
    "\n",
    "process_reports(reports, merge=True)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "213890f7-beed-48c1-9a36-843979da3c09",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- Loading /export/ml4ds/IntelComp/Datalake/CORDIS/20230823/parquet/reports.parquet\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Number of total reports: 51353\n",
      "-- Loading /export/ml4ds/IntelComp/Datalake/CORDIS/20230823/parquet/reports.parquet\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\t-- FP7: 21606 (42.07%)\n",
      "\t-- H2020: 29613 (57.67%)\n",
      "\t-- HORIZON: 134 (0.26%)\n"
     ]
    }
   ],
   "source": [
    "total = load_parquet(\"reports\").count()\n",
    "print(f\"Number of total reports: {total}\")\n",
    "\n",
    "for k, v in load_parquet(\"reports\").select(\"frameworkProgramme\").toPandas().groupby(\"frameworkProgramme\")[\"frameworkProgramme\"].count().items():\n",
    "        print(f\"\\t-- {k}: {v} ({v/total*100:.2f}%)\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0266d946",
   "metadata": {},
   "source": [
    "### Process EuroSciVoc Codes\n",
    "\n",
    "This part has been processed manually (at least partially) to get all the possible paths and codes, as no official reference has been found that includes all of them.\n",
    "\n",
    "**NOTE:**\n",
    "Some additional files not included in GitHub are required\n",
    "\n",
    "**NOTE2:**\n",
    "You do not need to run this cell, if you already have the EuroSciVoc files available"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "b82717f9",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "data/data_Pr_CORDIS/20220908/FP7/cordis-fp7projects-xlsx.zip\n",
      "-- Zip\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/lib/python3.8/site-packages/openpyxl/styles/stylesheet.py:226: UserWarning: Workbook contains no default style, apply openpyxl's default\n",
      "  warn(\"Workbook contains no default style, apply openpyxl's default\")\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "data/data_Pr_CORDIS/20220908/H2020/cordis-h2020projects-xlsx.zip\n",
      "-- Zip\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/lib/python3.8/site-packages/openpyxl/styles/stylesheet.py:226: UserWarning: Workbook contains no default style, apply openpyxl's default\n",
      "  warn(\"Workbook contains no default style, apply openpyxl's default\")\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "data/data_Pr_CORDIS/20220908/HORIZON/cordis-HORIZONprojects-xlsx.zip\n",
      "-- Zip\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/lib/python3.8/site-packages/openpyxl/styles/stylesheet.py:226: UserWarning: Workbook contains no default style, apply openpyxl's default\n",
      "  warn(\"Workbook contains no default style, apply openpyxl's default\")\n"
     ]
    }
   ],
   "source": [
    "import json\n",
    "\n",
    "with open(\"SciVoc-data/transform.json\", \"r\") as f:\n",
    "    taxonomies = json.load(f)\n",
    "\n",
    "# Get all available titles from SciVoc\n",
    "def process_keys(obj):\n",
    "    \"\"\"\n",
    "    Process json object to obtain all titles\n",
    "    \"\"\"\n",
    "    structure = []\n",
    "\n",
    "    def get_keys(obj):\n",
    "        keys = []\n",
    "        if isinstance(obj, list):\n",
    "            for el in obj:\n",
    "                keys.append(get_keys(el))\n",
    "        elif isinstance(obj, dict):\n",
    "            if obj.get(\"subtitles\"):\n",
    "                for el in obj[\"subtitles\"]:\n",
    "                    keys.extend([f\"{obj['title']}/{s.strip()}\" for s in get_keys(el)])\n",
    "            keys.append(obj[\"title\"])\n",
    "        return keys\n",
    "\n",
    "    # Convert obj to list\n",
    "    if not isinstance(obj, list):\n",
    "        obj = [obj]\n",
    "    for el in obj:\n",
    "        structure.extend(get_keys(el))\n",
    "    return structure\n",
    "\n",
    "\n",
    "all_keys = process_keys(taxonomies)\n",
    "all_keys = [f\"/{k}\" for k in all_keys]\n",
    "# Save\n",
    "with open(\"SciVoc-data/ScienceVocabulary.txt\", \"w\") as f:\n",
    "    [f.write(f\"{el}\\n\") for el in all_keys]\n",
    "\n",
    "# Load all EuroSciVoc\n",
    "dfs = []\n",
    "for f in projects:\n",
    "    print(f)\n",
    "    # Get type of file\n",
    "    extension = get_file_extension(f)\n",
    "    print(f\"-- {extension}\")\n",
    "\n",
    "    if extension == \"Zip\":\n",
    "        archive = zipfile.ZipFile(f.as_posix(), \"r\")\n",
    "        dfs.append(\n",
    "            pd.read_excel(\n",
    "                archive.read(\"xlsx/euroSciVoc.xlsx\"),\n",
    "                engine=\"openpyxl\",\n",
    "            )\n",
    "        )\n",
    "    elif extension == \"7-zip\":\n",
    "        with py7zr.SevenZipFile(f, \"r\") as archive:\n",
    "            dfs.append(\n",
    "                pd.read_excel(\n",
    "                    archive.read(\"xlsx/euroSciVoc.xlsx\")[\"xlsx/euroSciVoc.xlsx\"],\n",
    "                    engine=\"openpyxl\",\n",
    "                )\n",
    "            )\n",
    "# Concatenate both\n",
    "df = pd.concat(dfs)\n",
    "\n",
    "\n",
    "# Split codes to get only the code values\n",
    "df[\"euroSciVocCode\"] = df[\"euroSciVocCode\"].apply(\n",
    "    lambda x: [el.strip() for el in x.split(\"/\") if len(el) > 0]\n",
    ")\n",
    "df[\"euroSciVocPath\"] = df[\"euroSciVocPath\"].apply(\n",
    "    lambda x: [el.strip() for el in x.split(\"/\") if len(el) > 0]\n",
    ")\n",
    "df[\"euroSciVocTitle\"] = df[\"euroSciVocTitle\"].apply(str.strip)\n",
    "# Get length on the resulting lists\n",
    "df[\"length\"] = df[\"euroSciVocCode\"].apply(len)\n",
    "\n",
    "# Now we only want unique titles with longest codes (as this will be the entire path)\n",
    "euroSciVoc = (\n",
    "    df[[\"euroSciVocCode\", \"euroSciVocPath\", \"euroSciVocTitle\", \"length\"]]\n",
    "    .sort_values(by=\"length\")\n",
    "    .drop_duplicates(subset=\"euroSciVocTitle\", keep=\"last\")\n",
    ")\n",
    "\n",
    "# Also, as some elements are only found in the full path but don't have the code-title\n",
    "#  directly assigned we can obtain it from that path.\n",
    "aux = dict(\n",
    "    df.loc[\n",
    "        df.apply(lambda x: len(x[\"euroSciVocCode\"]) == len(x[\"euroSciVocPath\"]), axis=1)\n",
    "    ]\n",
    "    .apply(lambda x: list(zip(x[\"euroSciVocPath\"], x[\"euroSciVocCode\"])), axis=1)\n",
    "    .explode()\n",
    "    .drop_duplicates()\n",
    "    .tolist()\n",
    ")\n",
    "\n",
    "# Finally we get the title - code relationship\n",
    "title2code = dict(\n",
    "    euroSciVoc[[\"euroSciVocCode\", \"euroSciVocTitle\"]]\n",
    "    .apply(lambda x: (x[\"euroSciVocTitle\"], x[\"euroSciVocCode\"][-1]), axis=1)\n",
    "    .tolist()\n",
    ")\n",
    "title2code.update(aux)\n",
    "\n",
    "# Save all SciVoc information\n",
    "rows = []\n",
    "for k in all_keys:\n",
    "    this_row = []\n",
    "\n",
    "    spl_path = k.split(\"/\")\n",
    "\n",
    "    # Full path\n",
    "    this_row.append(k)\n",
    "\n",
    "    # Title\n",
    "    title = spl_path[-1]\n",
    "    this_row.append(title)\n",
    "\n",
    "    # Code\n",
    "    code = title2code.get(title, None)\n",
    "    this_row.append(code)\n",
    "\n",
    "    # Full code\n",
    "    full_code = \"/\".join([title2code.get(t, \"\") for t in spl_path])\n",
    "    this_row.append(full_code)\n",
    "\n",
    "    rows.append((this_row))\n",
    "scivoccodes = pd.DataFrame(rows, columns=[\"full_path\", \"title\", \"code\", \"full_code\"])\n",
    "scivoccodes.to_excel(\"SciVoc-data/SciVocCodes.xlsx\", index=None)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a2f9238c",
   "metadata": {},
   "source": [
    "### Organisations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "4426aeac-b7d8-4a0e-ace8-ca04e600ea8f",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "/export/data_ml4ds/IntelComp/Datasets/cordis/20230823/rawdata/FP7/cordis-fp7projects-xlsx.zip\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- Zip\n",
      "/export/data_ml4ds/IntelComp/Datasets/cordis/20230823/rawdata/H2020/cordis-h2020projects-xlsx.zip\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- Zip\n",
      "/export/data_ml4ds/IntelComp/Datasets/cordis/20230823/rawdata/HORIZON/cordis-HORIZONprojects-xlsx.zip\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- Zip\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/lib/python3.10/site-packages/openpyxl/styles/stylesheet.py:226: UserWarning: Workbook contains no default style, apply openpyxl's default\n",
      "  warn(\"Workbook contains no default style, apply openpyxl's default\")\n"
     ]
    }
   ],
   "source": [
    "# Load all organizations\n",
    "dfs = []\n",
    "for f in projects:\n",
    "    print(f)\n",
    "    # Get type of file\n",
    "    extension = get_file_extension(f)\n",
    "    print(f\"-- {extension}\")\n",
    "\n",
    "    if extension == \"Zip\":\n",
    "        archive = zipfile.ZipFile(f.as_posix(), \"r\")\n",
    "        dfs.append(\n",
    "            pd.read_excel(\n",
    "                archive.read(\"xlsx/organization.xlsx\"),\n",
    "                engine=\"openpyxl\",\n",
    "            )\n",
    "        )\n",
    "    elif extension == \"7-zip\":\n",
    "        with py7zr.SevenZipFile(f, \"r\") as archive:\n",
    "            dfs.append(\n",
    "                pd.read_excel(\n",
    "                    archive.read(\"xlsx/organization.xlsx\")[\"xlsx/organization.xlsx\"],\n",
    "                    engine=\"openpyxl\",\n",
    "                )\n",
    "            )\n",
    "df = pd.concat(dfs)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "55dc2aa9",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- Conversion OK\n",
      "-- Saving in /export/ml4ds/IntelComp/Datalake/CORDIS/20230823/parquet/organizations.parquet\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "23/08/26 23:30:32 WARN TaskSetManager: Stage 24 contains a task of very large size (1040 KiB). The maximum recommended task size is 1000 KiB.\n",
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SAVE SUCCESS\n"
     ]
    }
   ],
   "source": [
    "orgs_info = (\n",
    "    df[\n",
    "        [\n",
    "            \"organisationID\",\n",
    "            \"vatNumber\",\n",
    "            \"name\",\n",
    "            \"shortName\",\n",
    "            \"SME\",\n",
    "            \"activityType\",\n",
    "            \"street\",\n",
    "            \"postCode\",\n",
    "            \"city\",\n",
    "            \"country\",\n",
    "            \"nutsCode\",\n",
    "            \"geolocation\",\n",
    "            \"organizationURL\",\n",
    "            \"contentUpdateDate\",\n",
    "        ]\n",
    "    ]\n",
    "    .sort_values(by=[\"organisationID\", \"contentUpdateDate\"])\n",
    "    .drop_duplicates(subset=\"organisationID\", keep=\"last\")\n",
    "    .join(\n",
    "        df.groupby(\"organisationID\")[\"projectID\"].apply(list),\n",
    "        on=\"organisationID\",\n",
    "    )\n",
    ")\n",
    "\n",
    "sparkDF = convert_spark(orgs_info)\n",
    "save_parquet(sparkDF, \"organizations\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "27ce1c4d-9018-4974-a742-47c384a63ecd",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- Loading /export/ml4ds/IntelComp/Datalake/CORDIS/20230823/parquet/organizations.parquet\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "67348"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Count unique organizations according to id\n",
    "load_parquet(\"organizations\").select(\"organisationID\").distinct().count()\n",
    "\n",
    "# Save XLS file to NFS location\n",
    "# organizations_file = \"/export/data_ml4ds/IntelComp/Datasets/cordis/20220823/xlsx/organizations.xlsx\"\n",
    "# load_parquet(\"organizations\").toPandas().to_excel(organizations_file, index=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "30579042",
   "metadata": {},
   "source": [
    "### Projects"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "43b6d709",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- Microsoft\n",
      "Processing...\n",
      "\n",
      "-- Zip\n",
      "-- Files:\n",
      "-- -- xlsx/euroSciVoc.xlsx\n",
      "-- cordis-fp7projects-xlsx-euroSciVoc\n",
      "-- -- xlsx/legalBasis.xlsx\n",
      "-- cordis-fp7projects-xlsx-legalBasis\n",
      "-- -- xlsx/organization.xlsx\n",
      "-- cordis-fp7projects-xlsx-organization\n",
      "-- -- xlsx/project.xlsx\n",
      "-- cordis-fp7projects-xlsx-project\n",
      "-- -- xlsx/topics.xlsx\n",
      "-- cordis-fp7projects-xlsx-topics\n",
      "-- -- xlsx/webItem.xlsx\n",
      "-- cordis-fp7projects-xlsx-webItem\n",
      "-- -- xlsx/webLink.xlsx\n",
      "-- cordis-fp7projects-xlsx-webLink\n",
      "\n",
      " -------------------------------------------------------------------------------- \n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- Zip\n",
      "-- Files:\n",
      "-- -- xlsx/euroSciVoc.xlsx\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/lib/python3.10/site-packages/openpyxl/styles/stylesheet.py:226: UserWarning: Workbook contains no default style, apply openpyxl's default\n",
      "  warn(\"Workbook contains no default style, apply openpyxl's default\")\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- cordis-h2020projects-xlsx-euroSciVoc\n",
      "-- -- xlsx/legalBasis.xlsx\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/lib/python3.10/site-packages/openpyxl/styles/stylesheet.py:226: UserWarning: Workbook contains no default style, apply openpyxl's default\n",
      "  warn(\"Workbook contains no default style, apply openpyxl's default\")\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- cordis-h2020projects-xlsx-legalBasis\n",
      "-- -- xlsx/organization.xlsx\n",
      "-- cordis-h2020projects-xlsx-organization\n",
      "-- -- xlsx/project.xlsx\n",
      "-- cordis-h2020projects-xlsx-project\n",
      "-- -- xlsx/topics.xlsx\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/lib/python3.10/site-packages/openpyxl/styles/stylesheet.py:226: UserWarning: Workbook contains no default style, apply openpyxl's default\n",
      "  warn(\"Workbook contains no default style, apply openpyxl's default\")\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- cordis-h2020projects-xlsx-topics\n",
      "-- -- xlsx/webItem.xlsx\n",
      "-- cordis-h2020projects-xlsx-webItem\n",
      "-- -- xlsx/webLink.xlsx\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/lib/python3.10/site-packages/openpyxl/styles/stylesheet.py:226: UserWarning: Workbook contains no default style, apply openpyxl's default\n",
      "  warn(\"Workbook contains no default style, apply openpyxl's default\")\n",
      "/usr/lib/python3.10/site-packages/openpyxl/styles/stylesheet.py:226: UserWarning: Workbook contains no default style, apply openpyxl's default\n",
      "  warn(\"Workbook contains no default style, apply openpyxl's default\")\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- cordis-h2020projects-xlsx-webLink\n",
      "\n",
      " -------------------------------------------------------------------------------- \n",
      "\n",
      "-- Zip\n",
      "-- Files:\n",
      "-- -- xlsx/project.xlsx\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/compiler/lib/intel64/libiomp5.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_core.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "ERROR: ld.so: object '/opt/intel/composerxe/mkl/lib/intel64/libmkl_rt.so' from LD_PRELOAD cannot be preloaded (cannot open shared object file): ignored.\n",
      "/usr/lib/python3.10/site-packages/openpyxl/styles/stylesheet.py:226: UserWarning: Workbook contains no default style, apply openpyxl's default\n",
      "  warn(\"Workbook contains no default style, apply openpyxl's default\")\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- cordis-HORIZONprojects-xlsx-project\n",
      "-- -- xlsx/organization.xlsx\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/lib/python3.10/site-packages/openpyxl/styles/stylesheet.py:226: UserWarning: Workbook contains no default style, apply openpyxl's default\n",
      "  warn(\"Workbook contains no default style, apply openpyxl's default\")\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- cordis-HORIZONprojects-xlsx-organization\n",
      "-- -- xlsx/legalBasis.xlsx\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/lib/python3.10/site-packages/openpyxl/styles/stylesheet.py:226: UserWarning: Workbook contains no default style, apply openpyxl's default\n",
      "  warn(\"Workbook contains no default style, apply openpyxl's default\")\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- cordis-HORIZONprojects-xlsx-legalBasis\n",
      "-- -- xlsx/topics.xlsx\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/lib/python3.10/site-packages/openpyxl/styles/stylesheet.py:226: UserWarning: Workbook contains no default style, apply openpyxl's default\n",
      "  warn(\"Workbook contains no default style, apply openpyxl's default\")\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- cordis-HORIZONprojects-xlsx-topics\n",
      "-- -- xlsx/euroSciVoc.xlsx\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/lib/python3.10/site-packages/openpyxl/styles/stylesheet.py:226: UserWarning: Workbook contains no default style, apply openpyxl's default\n",
      "  warn(\"Workbook contains no default style, apply openpyxl's default\")\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- cordis-HORIZONprojects-xlsx-euroSciVoc\n",
      "-- -- xlsx/webLink.xlsx\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/lib/python3.10/site-packages/openpyxl/styles/stylesheet.py:226: UserWarning: Workbook contains no default style, apply openpyxl's default\n",
      "  warn(\"Workbook contains no default style, apply openpyxl's default\")\n",
      "/usr/lib/python3.10/site-packages/openpyxl/styles/stylesheet.py:226: UserWarning: Workbook contains no default style, apply openpyxl's default\n",
      "  warn(\"Workbook contains no default style, apply openpyxl's default\")\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- cordis-HORIZONprojects-xlsx-webLink\n",
      "-- -- xlsx/webItem.xlsx\n",
      "-- cordis-HORIZONprojects-xlsx-webItem\n",
      "\n",
      " -------------------------------------------------------------------------------- \n",
      "\n",
      "-- Conversion OK\n",
      "-- Loading /export/ml4ds/IntelComp/Datalake/CORDIS/20230823/parquet/publications.parquet\n",
      "LOADED publications\n",
      "-- Loading /export/ml4ds/IntelComp/Datalake/CORDIS/20230823/parquet/patents.parquet\n",
      "LOADED patents\n",
      "-- Saving in /export/ml4ds/IntelComp/Datalake/CORDIS/20230823/parquet/projects.parquet\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "23/08/26 23:51:56 WARN package: Truncated the string representation of a plan since it was too large. This behavior can be adjusted by setting 'spark.sql.debug.maxToStringFields'.\n",
      "23/08/26 23:51:56 WARN TaskSetManager: Stage 34 contains a task of very large size (2415 KiB). The maximum recommended task size is 1000 KiB.\n",
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SAVE SUCCESS\n"
     ]
    }
   ],
   "source": [
    "def process_projects(projects: List[Path]):\n",
    "    \"\"\"\n",
    "    Processall files given a list of Paths\n",
    "    Joins original project with topics, organizations, SciVoc, publications and patents\n",
    "    \"\"\"\n",
    "\n",
    "    projs = []\n",
    "\n",
    "    # Load SciVoc dictionary\n",
    "    _, scivoccodes = next(extract_file_info(Path(\"SciVoc-data/SciVocCodes.xlsx\")))\n",
    "    title2code = scivoccodes[[\"title\", \"code\"]].dropna()\n",
    "    title2code[\"code\"] = title2code[\"code\"].astype(int)\n",
    "    title2code[\"title\"] = title2code[\"title\"].apply(lambda x: get_only_letter(x.lower()))\n",
    "    title2code = dict(title2code.values)\n",
    "\n",
    "    print(\"Processing...\\n\")\n",
    "    for f in projects:\n",
    "\n",
    "        df_proj = None\n",
    "        tops = None\n",
    "        orgs = None\n",
    "        svcs = None\n",
    "\n",
    "        for fname, df in extract_file_info(f):\n",
    "            print(f\"-- {fname}\")\n",
    "\n",
    "            if fname is None or df is None:\n",
    "                continue\n",
    "\n",
    "            sfname = fname.split(\"-\")[-1]\n",
    "\n",
    "            # Drop:\n",
    "            if sfname in [\"legalBasis\", \"webItem\", \"webLink\"]:\n",
    "                continue\n",
    "\n",
    "            # Save\n",
    "            elif sfname == \"topics\":\n",
    "                # Merge entire dataframe with projects, as in this case each row is has unique (proj_id, topic, topic_title)\n",
    "                tops = (\n",
    "                    df[[\"topic\", \"projectID\", \"title\"]]\n",
    "                    .groupby(\"projectID\")[[\"topic\", \"title\"]]\n",
    "                    .apply(\n",
    "                        lambda x: pd.Series(\n",
    "                            {\n",
    "                                \"topic\": x[\"topic\"].values[0],\n",
    "                                \"topic_title\": x[\"title\"].values[0],\n",
    "                            }\n",
    "                        )\n",
    "                    )\n",
    "                )\n",
    "            elif sfname == \"organization\":\n",
    "                # Get information that is going to be added to projects\n",
    "                orgs = df[\n",
    "                    [\"projectID\", \"organisationID\", \"country\", \"role\", \"ecContribution\"]\n",
    "                ]\n",
    "                orgs.loc[:, \"ecContribution\"] = orgs.loc[:, \"ecContribution\"].apply(\n",
    "                    pd.to_numeric, errors=\"coerce\"\n",
    "                )\n",
    "                orgs.loc[pd.isna(orgs[\"ecContribution\"]), \"ecContribution\"] = 0.0\n",
    "                orgs = pd.DataFrame(\n",
    "                    orgs.groupby([\"projectID\"]).apply(\n",
    "                        lambda x: pd.Series(\n",
    "                            {\n",
    "                                \"countryContr\": \" \".join(\n",
    "                                    [\n",
    "                                        f\"{k}|{v}\"\n",
    "                                        for k, v in x.groupby(\"country\")[\n",
    "                                            \"ecContribution\"\n",
    "                                        ]\n",
    "                                        .apply(sum)\n",
    "                                        .items()\n",
    "                                    ]\n",
    "                                ),\n",
    "                                \"orgContr\": \" \".join(\n",
    "                                    [\n",
    "                                        f\"{i}|{c}\"\n",
    "                                        for i, c in zip(\n",
    "                                            x[\"organisationID\"].values,\n",
    "                                            x[\"ecContribution\"].values,\n",
    "                                        )\n",
    "                                    ]\n",
    "                                ),\n",
    "                                \"coordinatorCountry\": x.loc[\n",
    "                                    x[\"role\"] == \"coordinator\", \"country\"\n",
    "                                ].values[0],\n",
    "                                \"coordinatorOrg\": x.loc[\n",
    "                                    x[\"role\"] == \"coordinator\", \"organisationID\"\n",
    "                                ].values[0],\n",
    "                            }\n",
    "                        )\n",
    "                    )\n",
    "                )\n",
    "\n",
    "            elif sfname == \"euroSciVoc\":\n",
    "                svcs = (\n",
    "                    df[[\"euroSciVocTitle\", \"projectID\"]]\n",
    "                    .apply({\"euroSciVocTitle\":lambda x: get_only_letter(x.lower()),\n",
    "                            \"projectID\":lambda x: x})\n",
    "                    .groupby(\"projectID\")[\"euroSciVocTitle\"]\n",
    "                    .apply(lambda x: [title2code[i.strip()] for i in x.values if i.strip() in title2code.keys()])\n",
    "                    .rename(\"euroSciVocCode\")\n",
    "                )\n",
    "            elif sfname == \"project\":\n",
    "                df_proj = df.copy()\n",
    "            else:\n",
    "                continue\n",
    "\n",
    "        print(\"\\n\", \"-\" * 80, \"\\n\")\n",
    "\n",
    "        # Save enriched project\n",
    "        additions = pd.concat([tops, orgs, svcs], axis=1).reset_index(drop=False)\n",
    "        enrich_proj = (\n",
    "            df_proj.drop([\"legalBasis\", \"topics\"], axis=1)\n",
    "            .merge(additions, left_on=\"id\", right_on=\"projectID\", how=\"left\")\n",
    "            .drop(\"projectID\", axis=1)\n",
    "            .rename(columns={\"id\": \"projectID\"})\n",
    "        )\n",
    "        projs.append(enrich_proj)\n",
    "\n",
    "    # Conver to spark\n",
    "    projs_df = pd.concat(projs)\n",
    "    sparkDF = convert_spark(projs_df)\n",
    "\n",
    "    # Load publications\n",
    "    pubs = (\n",
    "        load_parquet(\"publications\")\n",
    "        .select(\"projectID\", \"id\")\n",
    "        .groupBy(\"projectID\")\n",
    "        .agg(F.collect_list(\"id\").alias(\"publicationID\"))\n",
    "    )\n",
    "    print(\"LOADED publications\")\n",
    "\n",
    "    # Load patents\n",
    "    pats = (\n",
    "        load_parquet(\"patents\")\n",
    "        .select(\"projectID\", \"patentFamilyIdentifier\")#\"appln_id\")\n",
    "        .groupBy(\"projectID\")\n",
    "        .agg(F.collect_list(\"patentFamilyIdentifier\").alias(\"patentID\"))\n",
    "    )\n",
    "    print(\"LOADED patents\")\n",
    "\n",
    "    sparkProjects = (\n",
    "        sparkDF\n",
    "        .join(pubs, on=\"projectID\", how=\"left\")\n",
    "        .join(pats, on=\"projectID\", how=\"left\")\n",
    "    )\n",
    "    save_parquet(sparkProjects, \"projects\")\n",
    "\n",
    "    return sparkProjects\n",
    "\n",
    "\n",
    "sparkProjects = process_projects(projects)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ec41e4a2",
   "metadata": {},
   "source": [
    "Some projects statistics:\n",
    "- Percentage of projects with publications\n",
    "- Percentage of projects with patents"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "590be03d",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- Loading /export/ml4ds/IntelComp/Datalake/CORDIS/20230823/parquet/projects.parquet\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Number of total projects: 69613\n",
      "\t-- FP7: 25785 (37.04%)\n",
      "\t-- H2020: 35386 (50.83%)\n",
      "\t-- HORIZON: 8442 (12.13%)\n",
      "Number of projects with publications: 37279 (53.55%)\n",
      "\t-- FP7: 15554 (22.34%)\n",
      "\t-- H2020: 21725 (31.21%)\n",
      "Number of projects with patents: 3034 (4.36%)\n",
      "\t-- FP7: 2155 (3.10%)\n",
      "\t-- H2020: 879 (1.26%)\n"
     ]
    }
   ],
   "source": [
    "def projects_statistics():\n",
    "    projs = load_parquet(\"projects\")\n",
    "    df = projs.select([\"frameworkProgramme\", \"publicationID\", \"patentID\"]).toPandas()\n",
    "    \n",
    "    total = len(df)\n",
    "    print(f\"Number of total projects: {total}\")\n",
    "    for k, v in df.groupby(\"frameworkProgramme\")[\"frameworkProgramme\"].count().items():\n",
    "        print(f\"\\t-- {k}: {v} ({v/total*100:.2f}%)\")\n",
    "\n",
    "    # Projects with publications\n",
    "    proj_pubs = df[[\"frameworkProgramme\", \"publicationID\"]].dropna(\n",
    "        subset=[\"publicationID\"]\n",
    "    )\n",
    "    num_pubs = proj_pubs[\"publicationID\"].count()\n",
    "    print(f\"Number of projects with publications: {num_pubs} ({num_pubs/total*100:.2f}%)\")\n",
    "    for k, v in (proj_pubs.groupby(\"frameworkProgramme\")[\"publicationID\"].count().items()):\n",
    "        print(f\"\\t-- {k}: {v} ({v/total*100:.2f}%)\")\n",
    "\n",
    "    # Projects with patents\n",
    "    proj_pats = df[[\"frameworkProgramme\", \"patentID\"]].dropna(subset=[\"patentID\"])\n",
    "    num_pats = proj_pats[\"patentID\"].count()\n",
    "    print(f\"Number of projects with patents: {num_pats} ({num_pats/total*100:.2f}%)\")\n",
    "    for k, v in proj_pats.groupby(\"frameworkProgramme\")[\"patentID\"].count().items():\n",
    "        print(f\"\\t-- {k}: {v} ({v/total*100:.2f}%)\")\n",
    "\n",
    "\n",
    "projects_statistics()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "id": "56503138-0e07-4a7e-9039-cf500e8ad5ad",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- Loading /export/ml4ds/IntelComp/Datalake/CORDIS/20220823/parquet/projects.parquet\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# Save XLS file to NFS location\n",
    "# projects_file = \"/export/data_ml4ds/IntelComp/Datasets/cordis/20220823/xlsx/projects.xlsx\"\n",
    "# load_parquet(\"projects\").toPandas().to_excel(projects_file, index=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "id": "b9b70283",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Processing hdfs://DTSCHDFSCluster/export/ml4ds/IntelComp/Datalake/CORDIS/20230823/parquet/organizations.parquet\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "67348\n",
      "root\n",
      " |-- organisationID: double (nullable = true)\n",
      " |-- vatNumber: string (nullable = true)\n",
      " |-- name: string (nullable = true)\n",
      " |-- shortName: string (nullable = true)\n",
      " |-- SME: boolean (nullable = true)\n",
      " |-- activityType: string (nullable = true)\n",
      " |-- street: string (nullable = true)\n",
      " |-- postCode: string (nullable = true)\n",
      " |-- city: string (nullable = true)\n",
      " |-- country: string (nullable = true)\n",
      " |-- nutsCode: string (nullable = true)\n",
      " |-- geolocation: string (nullable = true)\n",
      " |-- organizationURL: string (nullable = true)\n",
      " |-- contentUpdateDate: timestamp (nullable = true)\n",
      " |-- projectID: array (nullable = true)\n",
      " |    |-- element: long (containsNull = true)\n",
      "\n",
      "+--------------+-----------+--------------------+----------------+-----+------------+--------------------+--------+--------------------+-------+--------+--------------------+---------------+-------------------+--------------------+\n",
      "|organisationID|  vatNumber|                name|       shortName|  SME|activityType|              street|postCode|                city|country|nutsCode|         geolocation|organizationURL|  contentUpdateDate|           projectID|\n",
      "+--------------+-----------+--------------------+----------------+-----+------------+--------------------+--------+--------------------+-------+--------+--------------------+---------------+-------------------+--------------------+\n",
      "|  9.99785403E8|       null|HUTCHISON WHAMPOA...|            HWEL| null|         PRC|HUTCHISON HOUSE 5...|SW11 4AN|              LONDON|     UK|    null|                null|           null|2022-09-03 22:40:26|            [216462]|\n",
      "|  9.99785694E8|       null| FUNDACION BARILOCHE|              FB| null|         PRC|           SUIZA 970|    8400|SAN CARLOS DE BAR...|     AR|    null|                null|           null|2019-07-15 16:08:24|            [213320]|\n",
      "|  9.99785791E8|BG123047654|CHAMBER OF COMMER...|CCI Stara Zagora|false|         OTH|G RAKOVSKI STREET 66|    6000|        STARA ZAGORA|     BG|    null|42.4247421,25.625...|           null|2022-10-28 14:10:00|[243481, 830902, ...|\n",
      "|  9.99785888E8|       null|          Q-Free ASA|          Q-FREE| null|         PRC|THONNING OWESENSG...|    7044|           TRONDHEIM|     NO|    null|                null|           null|2022-05-25 23:08:46|[317547, 218354, ...|\n",
      "|  9.99785985E8|       null|ETEL AUSTRIA GMBH...|              TA| null|         PRC|   LASSALLESTRASSE 9|    1020|                WIEN|     AT|    null|                null|           null|2019-07-16 11:25:11|            [216556]|\n",
      "+--------------+-----------+--------------------+----------------+-----+------------+--------------------+--------+--------------------+-------+--------+--------------------+---------------+-------------------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "\n",
      " -------------------------------------------------------------------------------- \n",
      "\n",
      "Processing hdfs://DTSCHDFSCluster/export/ml4ds/IntelComp/Datalake/CORDIS/20230823/parquet/patents.parquet\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "32073\n",
      "root\n",
      " |-- projectID: long (nullable = true)\n",
      " |-- type: string (nullable = true)\n",
      " |-- title: string (nullable = true)\n",
      " |-- organisationID: double (nullable = true)\n",
      " |-- applicantName: string (nullable = true)\n",
      " |-- applicationDate: timestamp (nullable = true)\n",
      " |-- applicationPrefix: string (nullable = true)\n",
      " |-- applicationIdentifier: string (nullable = true)\n",
      " |-- applicationKind: string (nullable = true)\n",
      " |-- epoAppUrl: string (nullable = true)\n",
      " |-- priorityDate: timestamp (nullable = true)\n",
      " |-- patentType: string (nullable = true)\n",
      " |-- awardDate: timestamp (nullable = true)\n",
      " |-- awardPrefix: string (nullable = true)\n",
      " |-- awardIdentifier: string (nullable = true)\n",
      " |-- awardKind: string (nullable = true)\n",
      " |-- epoPubUrl: string (nullable = true)\n",
      " |-- patentFamilyIdentifier: long (nullable = true)\n",
      " |-- frameworkProgramme: string (nullable = true)\n",
      " |-- appln_id: string (nullable = true)\n",
      "\n",
      "+---------+------+--------------------+--------------+--------------------+-------------------+-----------------+---------------------+---------------+--------------------+-------------------+----------+-------------------+-----------+---------------+---------+--------------------+----------------------+------------------+---------+\n",
      "|projectID|  type|               title|organisationID|       applicantName|    applicationDate|applicationPrefix|applicationIdentifier|applicationKind|           epoAppUrl|       priorityDate|patentType|          awardDate|awardPrefix|awardIdentifier|awardKind|           epoPubUrl|patentFamilyIdentifier|frameworkProgramme| appln_id|\n",
      "+---------+------+--------------------+--------------+--------------------+-------------------+-----------------+---------------------+---------------+--------------------+-------------------+----------+-------------------+-----------+---------------+---------+--------------------+----------------------+------------------+---------+\n",
      "|   201862|PATENT|P140Cap in tumour...|  9.99861936E8|Paola Defilippi, ...|2006-09-08 00:00:00|               EP|             06795477|              A|https://worldwide...|2006-09-08 00:00:00|BACKGROUND|2009-05-27 00:00:00|         EP|        2061494|       A1|https://worldwide...|              37969726|               FP7| 56192422|\n",
      "|   201862|PATENT|P140Cap in tumour...|  9.99861936E8|Paola Defilippi, ...|2006-09-08 00:00:00|               EP|             06795477|              A|https://worldwide...|2006-09-08 00:00:00|BACKGROUND|2009-05-27 00:00:00|         EP|        2061494|       A1|https://worldwide...|              37969726|               FP7|341492391|\n",
      "|   286552|PATENT|Ultrasonic device...|  9.98473575E8|Ascamm Technology...|2008-08-08 00:00:00|               DK|             08828843|              T|https://worldwide...|2007-08-09 00:00:00|BACKGROUND|2012-07-23 00:00:00|         DK|        2189264|       T3|https://worldwide...|              40386727|               FP7|365299500|\n",
      "|   286552|PATENT|Ultrasonic device...|  9.98473575E8|Ascamm Technology...|2008-08-08 00:00:00|               DK|             08828843|              T|https://worldwide...|2007-08-09 00:00:00|BACKGROUND|2012-07-23 00:00:00|         DK|        2189264|       T3|https://worldwide...|              40386727|               FP7|365959789|\n",
      "|   286552|PATENT|Ultrasonic device...|  9.98473575E8|Ascamm Technology...|2008-08-08 00:00:00|               DK|             08828843|              T|https://worldwide...|2007-08-09 00:00:00|BACKGROUND|2012-07-23 00:00:00|         DK|        2189264|       T3|https://worldwide...|              40386727|               FP7|523313873|\n",
      "+---------+------+--------------------+--------------+--------------------+-------------------+-----------------+---------------------+---------------+--------------------+-------------------+----------+-------------------+-----------+---------------+---------+--------------------+----------------------+------------------+---------+\n",
      "only showing top 5 rows\n",
      "\n",
      "\n",
      " -------------------------------------------------------------------------------- \n",
      "\n",
      "Processing hdfs://DTSCHDFSCluster/export/ml4ds/IntelComp/Datalake/CORDIS/20230823/parquet/projects.parquet\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "69613\n",
      "root\n",
      " |-- projectID: long (nullable = true)\n",
      " |-- acronym: string (nullable = true)\n",
      " |-- status: string (nullable = true)\n",
      " |-- title: string (nullable = true)\n",
      " |-- startDate: timestamp (nullable = true)\n",
      " |-- endDate: timestamp (nullable = true)\n",
      " |-- totalCost: string (nullable = true)\n",
      " |-- ecMaxContribution: string (nullable = true)\n",
      " |-- ecSignatureDate: timestamp (nullable = true)\n",
      " |-- frameworkProgramme: string (nullable = true)\n",
      " |-- masterCall: string (nullable = true)\n",
      " |-- subCall: string (nullable = true)\n",
      " |-- fundingScheme: string (nullable = true)\n",
      " |-- nature: string (nullable = true)\n",
      " |-- objective: string (nullable = true)\n",
      " |-- contentUpdateDate: timestamp (nullable = true)\n",
      " |-- rcn: long (nullable = true)\n",
      " |-- grantDoi: string (nullable = true)\n",
      " |-- topic: string (nullable = true)\n",
      " |-- topic_title: string (nullable = true)\n",
      " |-- countryContr: string (nullable = true)\n",
      " |-- orgContr: string (nullable = true)\n",
      " |-- coordinatorCountry: string (nullable = true)\n",
      " |-- coordinatorOrg: double (nullable = true)\n",
      " |-- euroSciVocCode: array (nullable = true)\n",
      " |    |-- element: long (containsNull = true)\n",
      " |-- publicationID: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- patentID: array (nullable = true)\n",
      " |    |-- element: long (containsNull = true)\n",
      "\n",
      "+---------+-----------+------+--------------------+-------------------+-------------------+------------+-----------------+-------------------+------------------+--------------------+--------------------+--------------+------+--------------------+-------------------+------+--------------+--------------------+--------------------+--------------------+--------------------+------------------+--------------+--------------------+--------------------+--------+\n",
      "|projectID|    acronym|status|               title|          startDate|            endDate|   totalCost|ecMaxContribution|    ecSignatureDate|frameworkProgramme|          masterCall|             subCall| fundingScheme|nature|           objective|  contentUpdateDate|   rcn|      grantDoi|               topic|         topic_title|        countryContr|            orgContr|coordinatorCountry|coordinatorOrg|      euroSciVocCode|       publicationID|patentID|\n",
      "+---------+-----------+------+--------------------+-------------------+-------------------+------------+-----------------+-------------------+------------------+--------------------+--------------------+--------------+------+--------------------+-------------------+------+--------------+--------------------+--------------------+--------------------+--------------------+------------------+--------------+--------------------+--------------------+--------+\n",
      "|   100017|      SOFIA|   ONG|Smart Objects For...|2009-01-01 00:00:00|2012-03-31 00:00:00|3.62883491E7|        6060154.3|               null|               FP7|                null|      ARTEMIS-2008-1|JTI-CP-ARTEMIS|  null|\"The mission of t...|2022-02-23 15:24:10|106175|          null|SP1-JTI-ARTEMIS-2...|Smart environment...|DE|56003.45 ES|78...|996613600.0|56182...|                FI|  9.99912085E8|    [1269, 64785222]|                null|    null|\n",
      "|   115009| PHARMA-COG|   ONG|Prediction of cog...|2010-01-01 00:00:00|2015-12-31 00:00:00| 3.0715556E7|        9658388.0|               null|               FP7|                null|      IMI-JU-01-2008|    JTI-CP-IMI|  null|PHARMA-COG aims t...|2023-04-05 12:40:17|203681|          null|   IMI-JU-01-2008-11|Neurodegenerative...|BE|0.0 CH|0.0 DE|...|999838074.0|36000...|                UK|  9.99939633E8|[79909316, 1007, ...|[C5BD8110501C1759...|    null|\n",
      "|   116060|    IMPRiND|SIGNED|Inhibiting Misfol...|2017-03-01 00:00:00|2022-02-28 00:00:00| 1.1388398E7|        4684998.0|2017-04-07 00:00:00|             H2020|H2020-JTI-IMI2-20...|H2020-JTI-IMI2-20...|           RIA|  null|Assemblies of tau...|2022-12-31 18:29:34|209883|10.3030/116060|     IMI2-2015-07-02|IDENTIFICATION OF...|BE|420000.0 CH|20...|999997736|208000....|                UK|   9.9998435E8|[633, 1371, 653, ...|[116060_1317980_P...|    null|\n",
      "|   200228|R3SEARCHERS|   CLO|Discover the Rese...|2007-06-01 00:00:00|2007-12-31 00:00:00|    102932.8|         72939.55|               null|               FP7|                null|FP7-PEOPLE-2007-5...|        CSA-SA|  null|\"The R3searchers ...|2022-09-03 22:39:56| 90049|          null|PEOPLE-2007-5-1.N...|Marie Curie Actio...|         ES|72939.55|999867077.0|20313...|                ES|  9.99521272E8|                null|                null|    null|\n",
      "|   200238|    IDFFERN|   CLO|ILE DE FRANCE FRE...|2007-06-01 00:00:00|2008-02-29 00:00:00|     86412.0|          55000.0|               null|               FP7|                null|FP7-PEOPLE-2007-5...|        CSA-SA|  null|Researchers’ Nigh...|2022-09-03 22:38:43| 90053|          null|PEOPLE-2007-5-1.N...|Marie Curie Actio...|          FR|55000.0|999966114.0|1450....|                FR|  9.99551051E8|                null|                null|    null|\n",
      "+---------+-----------+------+--------------------+-------------------+-------------------+------------+-----------------+-------------------+------------------+--------------------+--------------------+--------------+------+--------------------+-------------------+------+--------------+--------------------+--------------------+--------------------+--------------------+------------------+--------------+--------------------+--------------------+--------+\n",
      "only showing top 5 rows\n",
      "\n",
      "\n",
      " -------------------------------------------------------------------------------- \n",
      "\n",
      "Processing hdfs://DTSCHDFSCluster/export/ml4ds/IntelComp/Datalake/CORDIS/20230823/parquet/publications.parquet\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "731356\n",
      "root\n",
      " |-- projectID: long (nullable = true)\n",
      " |-- title: string (nullable = true)\n",
      " |-- authors: string (nullable = true)\n",
      " |-- isPublishedAs: string (nullable = true)\n",
      " |-- repositoryUrl: string (nullable = true)\n",
      " |-- journalTitle: string (nullable = true)\n",
      " |-- publisher: string (nullable = true)\n",
      " |-- journalNumber: string (nullable = true)\n",
      " |-- publishedPages: string (nullable = true)\n",
      " |-- doi: string (nullable = true)\n",
      " |-- id: string (nullable = true)\n",
      " |-- frameworkProgramme: string (nullable = true)\n",
      " |-- publishedYear: double (nullable = true)\n",
      " |-- issn: string (nullable = true)\n",
      " |-- isbn: string (nullable = true)\n",
      " |-- projectAcronym: string (nullable = true)\n",
      " |-- collection: string (nullable = true)\n",
      " |-- contentUpdateDate: timestamp (nullable = true)\n",
      " |-- rcn: double (nullable = true)\n",
      " |-- SSID: long (nullable = true)\n",
      "\n",
      "+---------+--------------------+-------------------+--------------------+-------------+--------------------+---------+-------------+--------------+----+--------------------+------------------+-------------+----+----+--------------+----------+-----------------+----+----+\n",
      "|projectID|               title|            authors|       isPublishedAs|repositoryUrl|        journalTitle|publisher|journalNumber|publishedPages| doi|                  id|frameworkProgramme|publishedYear|issn|isbn|projectAcronym|collection|contentUpdateDate| rcn|SSID|\n",
      "+---------+--------------------+-------------------+--------------------+-------------+--------------------+---------+-------------+--------------+----+--------------------+------------------+-------------+----+----+--------------+----------+-----------------+----+----+\n",
      "|   226354|Skeneimorph speci...| Hoffman, L. et al.|                null|         null|Miscellenea Malac...|     null|            4|         47-61|null|AE0DC2215ADBD2384...|               FP7|         null|null|null|          null|      null|             null|null|null|\n",
      "|   226354|Margarites toroid...|Hoffmann, L. et al.|                null|         null|Miscellanea Malac...|     null|            5|         57-60|null|C0580064A23994494...|               FP7|         null|null|null|          null|      null|             null|null|null|\n",
      "|   226354|Yoldiella ovulum ...|Hoffmann, L. et al.|                null|         null|Miscellanea Malac...|     null|            5|         53-56|null|7F37E0659EE7FE0FF...|               FP7|         null|null|null|          null|      null|             null|null|null|\n",
      "|   226354|Gastropoda (Mollu...|Hoffmann, L. et al.|                null|         null|Miscellenea Malac...|     null|            4|        85-118|null|5C60AD6EA275A36DD...|               FP7|         null|null|null|          null|      null|             null|null|null|\n",
      "|   226354|Coralliophilinae ...| Taviani, M. et al.|PEER REVIEWED ART...|         null|            Nautilus|     null|          123|       106-112|null|A812CAF5E41B9F26E...|               FP7|         null|null|null|          null|      null|             null|null|null|\n",
      "+---------+--------------------+-------------------+--------------------+-------------+--------------------+---------+-------------+--------------+----+--------------------+------------------+-------------+----+----+--------------+----------+-----------------+----+----+\n",
      "only showing top 5 rows\n",
      "\n",
      "\n",
      " -------------------------------------------------------------------------------- \n",
      "\n",
      "Processing hdfs://DTSCHDFSCluster/export/ml4ds/IntelComp/Datalake/CORDIS/20230823/parquet/reports.parquet\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 65:======================================================> (39 + 0) / 40]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "51353\n",
      "root\n",
      " |-- id: string (nullable = true)\n",
      " |-- title: string (nullable = true)\n",
      " |-- projectID: long (nullable = true)\n",
      " |-- projectAcronym: string (nullable = true)\n",
      " |-- attachment: string (nullable = true)\n",
      " |-- contentUpdateDate: timestamp (nullable = true)\n",
      " |-- rcn: long (nullable = true)\n",
      " |-- frameworkProgramme: string (nullable = true)\n",
      "\n",
      "+------------+--------------------+---------+--------------+--------------------+-------------------+------+------------------+\n",
      "|          id|               title|projectID|projectAcronym|          attachment|  contentUpdateDate|   rcn|frameworkProgramme|\n",
      "+------------+--------------------+---------+--------------+--------------------+-------------------+------+------------------+\n",
      "|   892354_PS|Periodic Reportin...|   892354|EXCHANGE_inLCs|/docs/results/h20...|2023-04-16 11:54:09|928093|             H2020|\n",
      "|101002685_PS|Periodic Reportin...|101002685|        ARTIST|/docs/results/h20...|2023-04-16 11:52:51|928059|             H2020|\n",
      "|101006667_PS|Periodic Reportin...|101006667|       SO-FREE|/docs/results/h20...|2023-04-16 12:18:04|928505|             H2020|\n",
      "|   792600_PS|Periodic Reportin...|   792600|  SCIENCE-MINQ|/docs/results/h20...|2023-04-16 12:17:53|928496|             H2020|\n",
      "|   893196_PS|Periodic Reportin...|   893196|       TGF-BTB|/docs/results/h20...|2023-04-16 12:17:26|928486|             H2020|\n",
      "+------------+--------------------+---------+--------------+--------------------+-------------------+------+------------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "\n",
      " -------------------------------------------------------------------------------- \n",
      "\n",
      "CPU times: user 30.2 ms, sys: 8.02 ms, total: 38.2 ms\n",
      "Wall time: 26.9 s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "hdfs = True\n",
    "if hdfs:\n",
    "    # Files in HDFS\n",
    "    hdfs_dir_raw = spark._jvm.org.apache.hadoop.fs.Path(dir_parquet.as_posix())\n",
    "    hdfs_file_list = [el.getPath().toUri().toString() for el in fs.listStatus(hdfs_dir_raw)]\n",
    "\n",
    "    for f in hdfs_file_list:\n",
    "        if f.endswith(\".parquet\"):\n",
    "            print(f\"Processing {f}\")\n",
    "            sparkDF = spark.read.parquet(f)\n",
    "            print(sparkDF.count())\n",
    "            sparkDF.printSchema()\n",
    "            sparkDF.show(5)\n",
    "            print(\"\\n\", \"-\" * 80, \"\\n\")\n",
    "else:\n",
    "    for f in dir_parquet.iterdir():\n",
    "        if f.name.endswith(\".parquet\"):\n",
    "            print(f\"Processing {f}\")\n",
    "            sparkDF = spark.read.parquet(f.as_posix())\n",
    "            print(sparkDF.count())\n",
    "            sparkDF.printSchema()\n",
    "            sparkDF.show(5)\n",
    "            print(\"\\n\", \"-\" * 80, \"\\n\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "88f3b749-cda5-4dad-b80b-4362cdad0496",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.10.12"
  },
  "vscode": {
   "interpreter": {
    "hash": "7c166760bc9517b789c32d5e4045207aa3c23cd3b35b3de2e2f89ee000e58fd0"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}