{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.10.13","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"none","dataSources":[{"sourceId":56537,"databundleVersionId":8015876,"sourceType":"competition"}],"dockerImageVersionId":30698,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"# This Python 3 environment comes with many helpful analytics libraries installed\n# It is defined by the kaggle/python Docker image: https://github.com/kaggle/docker-python\n# For example, here's several helpful packages to load\n\n# import numpy as np # linear algebra\n# import pandas as pd # data processing, CSV file I/O (e.g. pd.read_csv)\n\n# Input data files are available in the read-only \"../input/\" directory\n# For example, running this (by clicking run or pressing Shift+Enter) will list all files under the input directory\n\nimport os\nfor dirname, _, filenames in os.walk('/kaggle/input'):\n    for filename in filenames:\n        print(os.path.join(dirname, filename))\n\n# You can write up to 20GB to the current directory (/kaggle/working/) that gets preserved as output when you create a version using \"Save & Run All\" \n# You can also write temporary files to /kaggle/temp/, but they won't be saved outside of the current session","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"from kaggle_secrets import UserSecretsClient\nuser_secrets = UserSecretsClient()\nsecret_value_0 = user_secrets.get_secret(\"KAGGLE_KEY\")\nsecret_value_1 = user_secrets.get_secret(\"KAGGLE_USERNAME\")\n\n# Set Kaggle authentication credentials\nos.environ['KAGGLE_USERNAME'] = secret_value_1\nos.environ['KAGGLE_KEY'] = secret_value_0","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# import os\nimport json\nimport pandas as pd\nimport pyarrow as pa\nimport pyarrow.parquet as pq\nfrom kaggle.api.kaggle_api_extended import KaggleApi\n\n# Parameters\ninput_file = \"/kaggle/input/leap-atmospheric-physics-ai-climsim/train.csv\"  # Input file path\nchunk_size = 100_000  # Number of rows to process per chunk\nstorage_limit = 18  # Storage limit in GB\ndataset_title_prefix = \"LEAP Parquet Dataset\"  # Dataset title prefix\ndataset_slug_prefix = dataset_title_prefix.lower().replace(\" \", \"-\")  # Dataset slug prefix\n\n# Authenticate Kaggle API\napi = KaggleApi()\napi.authenticate()\n\ndef upload_dataset(dataset_number, dataset_dir):\n    \"\"\"\n    Upload a dataset to Kaggle.\n    \n    Args:\n        dataset_number (int): The dataset number.\n        dataset_dir (str): The directory containing the dataset files.\n        \n    Returns:\n        str: The slug of the created dataset.\n    \"\"\"\n    dataset_title = f\"{dataset_title_prefix} - Part {dataset_number}\"\n    dataset_slug = f\"{dataset_slug_prefix}-part-{dataset_number}\"\n\n    # Create a new dataset on Kaggle\n    metadata_path = f'{dataset_dir}/dataset-metadata.json'\n    dataset_metadata = {\n       \"title\": dataset_title,\n       \"id\": f\"{api.config_values['username']}/{dataset_slug}\",\n       \"licenses\": [{\"name\": \"CC0-1.0\"}]\n    }\n    with open(metadata_path, 'w') as f:\n        json.dump(dataset_metadata, f, indent=4)\n    with open(metadata_path, 'r') as f:\n        print(f.read())  # Check the contents of the metadata\n    api.dataset_create_new(folder=dataset_dir, public=False, convert_to_csv=False, dir_mode='tar')\n    print(f\"Created new dataset: {dataset_slug}\")\n    return dataset_slug\n\ndef process_chunk(i, chunk, dataset_dir):\n    \"\"\"\n    Process a chunk of data and save it as a Parquet file.\n    \n    Args:\n        i (int): The chunk index.\n        chunk (pandas.DataFrame): The chunk of data to process.\n        dataset_dir (str): The directory to save the Parquet file.\n        \n    Returns:\n        tuple: A tuple containing the updated dataset directory and the total output folder size.\n    \"\"\"\n    table = pa.Table.from_pandas(chunk)\n    parquet_file_path = f'{dataset_dir}/train_{i:03d}.parquet'\n    pq.write_table(table, parquet_file_path, compression='zstd', use_dictionary=False)\n\n    parquet_file_size = os.path.getsize(parquet_file_path)\n    print(f\"Processed chunk {i} and saved as {parquet_file_path} (Size: {parquet_file_size/1024**3:.2f} GB)\")\n\n    output_folder_size = get_output_folder_size(dataset_dir)\n    print(f\"Current output folder size: {output_folder_size/1024**3:.2f} GB\")\n\n    if output_folder_size > storage_limit * 1024**3:\n        dataset_number = int(dataset_dir.split(\"-\")[-1])\n        print(f\"Uploading dataset: {dataset_dir}\")\n        dataset_slug = upload_dataset(dataset_number, dataset_dir)\n\n        # Remove uploaded Parquet files from the output folder\n        print(\"Removing uploaded Parquet files from the output folder\")\n        for file in os.listdir(dataset_dir):\n            file_path = os.path.join(dataset_dir, file)\n            if os.path.isfile(file_path) and file.endswith(\".parquet\"):\n                os.remove(file_path)\n        print(\"Removed\")\n\n        dataset_number += 1\n        dataset_dir = f\"{dataset_slug_prefix}-part-{dataset_number}\"\n        os.makedirs(dataset_dir, exist_ok=True)\n\n    return dataset_dir, output_folder_size + parquet_file_size\n\ndef get_output_folder_size(dataset_dir):\n    \"\"\"\n    Calculate the total size of files in the output folder.\n    \n    Args:\n        dataset_dir (str): The directory containing the dataset files.\n        \n    Returns:\n        int: The total size of files in bytes.\n    \"\"\"\n    total_size = sum(os.path.getsize(os.path.join(dataset_dir, f)) for f in os.listdir(dataset_dir) if os.path.isfile(os.path.join(dataset_dir, f)))\n    return total_size\n\n# Initialize dataset variables\ndataset_number = 1\ndataset_dir = f\"{dataset_slug_prefix}-part-{dataset_number}\"\nos.makedirs(dataset_dir, exist_ok=True)\noutput_folder_size = 0\n\n# Read the CSV file sequentially and convert to Parquet files\nprint(\"Start reading the CSV file\")\nfor i, chunk in enumerate(pd.read_csv(input_file, chunksize=chunk_size)):\n    dataset_dir, output_folder_size = process_chunk(i, chunk, dataset_dir)\n\n# Upload the final dataset\nprint(f\"Uploading final dataset: {dataset_dir}\")\ndataset_slug = upload_dataset(dataset_number, dataset_dir)\n\n# Remove uploaded Parquet files from the output folder\nfor file in os.listdir(dataset_dir):\n    file_path = os.path.join(dataset_dir, file)\n    if os.path.isfile(file_path) and file.endswith(\".parquet\"):\n        os.remove(file_path)\n\nprint(\"CSV to Parquet conversion and upload completed.\")","metadata":{"trusted":true},"execution_count":null,"outputs":[]}]}