spark.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.conf import SparkConf\n",
    "from pyspark.context import SparkContext\n",
    "\n",
    "#   https://kashif-sohail.medium.com/read-files-from-google-cloud-storage-bucket-using-local-pyspark-and-jupyter-notebooks-f8bd43f4b42e"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Build Master"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "24/04/12 16:36:02 WARN Utils: Your hostname, manuelpc resolves to a loopback address: 127.0.0.1; using 131.155.193.147 instead (on interface wlo1)\n",
      "24/04/12 16:36:02 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address\n",
      "24/04/12 16:36:03 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n"
     ]
    }
   ],
   "source": [
    "keyfile = \"/home/manuel/Projects/DataEngineeringZoomcamp/infra/creds.json\"\n",
    "\n",
    "# Use spark gcs connector\n",
    "conf = (SparkConf()\n",
    "        .setMaster(\"spark://localhost:7077\")\n",
    "        .setAppName(\"GCSRead\")\n",
    "        .set(\"spark.jars\", \"./spark-jars/gcs-connector-hadoop3-latest.jar,./spark-jars/spark-3.5-bigquery-0.37.0.jar\")\n",
    "        .set(\"spark.hadoop.google.cloud.auth.service.account.enable\", \"true\")\n",
    "        .set(\"spark.hadoop.google.cloud.auth.service.account.json.keyfile\", keyfile)\n",
    "        .set(\"spark.sql.legacy.parquet.nanosAsLong\", \"true\")\n",
    ")\n",
    "\n",
    "sc = SparkContext(conf=conf)\n",
    "\n",
    "hadoop_conf = sc._jsc.hadoopConfiguration()\n",
    "\n",
    "hadoop_conf.set(\"fs.gs.impl\", \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem\")\n",
    "hadoop_conf.set(\"fs.AbstractFileSystem.gs.impl\", \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS\")\n",
    "hadoop_conf.set(\"google.cloud.auth.service.account.enable\", \"true\")\n",
    "hadoop_conf.set(\"google.cloud.auth.service.account.json.keyfile\", keyfile)\n",
    "\n",
    "spark: SparkSession = SparkSession.builder.config(conf=sc.getConf()).getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Read DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "24/04/12 16:36:17 WARN GarbageCollectionMetrics: To enable non-built-in garbage collector(s) List(G1 Concurrent GC), users should configure it(them) to spark.eventLog.gcMetrics.youngGenerationGarbageCollectors or spark.eventLog.gcMetrics.oldGenerationGarbageCollectors\n",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "path=f\"gs://weather_data_de_bucket/Actuele10mindataKNMIstations/2/2024/04/*/*/*/weather_data.parquet\"\n",
    "\n",
    "df = spark.read.parquet(path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "332"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "24/04/12 16:36:44 WARN SparkStringUtils: Truncated the string representation of a plan since it was too large. This behavior can be adjusted by setting 'spark.sql.debug.maxToStringFields'.\n",
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------+-----------+-------+------+------+----+-----+-----+----+---+-----+-------+-----+----+----+----+-------+-----+-------+-------+----+----+-------+-------+----+----+----+----+----+----+----+----+----+----+-------+-------+---+---+------+---+----+----+----+----+----+-------+----+----+----+----+----+----+----+-------+-------+-------+-------+---+----+-------+-------+-------+----+----+-------+----+-----+-----+----+-----+-----+----+----+----+----+----+----+----+-----+-----+----+----+----+----+----+----+----+----+----+----+-------+---+------+---+-----+-------+-----------+-------+----------+---+---+-------+-------------------+\n",
      "|            wsi|stationname|    lat|   lon|height| D1H|   dd|   dn|  dr|dsd|   dx|     ff|  ffs| fsd|  fx| fxs|    gff| gffs|      h|     h1|  h2|  h3|     hc|    hc1| hc2| hc3|   n|  n1|  n2|  n3|  nc| nc1| nc2| nc3|     p0|     pp| pg| pr|    ps|pwc| Q1H|Q24H|  qg| qgn| qgx|    qnh|R12H| R1H|R24H| R6H|  rg|  rh|rh10|  Sav1H|  Sax1H|  Sax3H|  Sax6H| sq|  ss|   Sx1H|   Sx3H|   Sx6H| t10|  ta|     tb| tb1|Tb1n6|Tb1x6| tb2|Tb2n6|Tb2x6| tb3| tb4| tb5|  td|td10|  tg| tgn|Tgn12|Tgn14|Tgn6|  tn|Tn12|Tn14| Tn6| tsd|  tx|Tx12|Tx24| Tx6|     vv|W10|W10-10| ww|ww-10|     zm|iso_dataset|product|projection|nhc| za|station|               time|\n",
      "+---------------+-----------+-------+------+------+----+-----+-----+----+---+-----+-------+-----+----+----+----+-------+-----+-------+-------+----+----+-------+-------+----+----+----+----+----+----+----+----+----+----+-------+-------+---+---+------+---+----+----+----+----+----+-------+----+----+----+----+----+----+----+-------+-------+-------+-------+---+----+-------+-------+-------+----+----+-------+----+-----+-----+----+-----+-----+----+----+----+----+----+----+----+-----+-----+----+----+----+----+----+----+----+----+----+----+-------+---+------+---+-----+-------+-----------+-------+----------+---+---+-------+-------------------+\n",
      "|0-20000-0-06201|   D15-FA-1|54.3256|2.9358|  42.7|NULL|233.1|205.3|NULL|4.3|251.7|8.47339|10.25|0.22|NULL|NULL|9.43676|11.01|31434.9|31434.9| 0.0| 0.0|31434.9|31434.9| 0.0| 0.0| 1.0| 1.0| 0.0| 0.0| 1.0| 1.0| 0.0| 0.0|1021.13|1026.38|0.0|0.0|1021.9|0.0|NULL|NULL|NULL|NULL|NULL|1026.26|NULL|NULL|NULL|NULL|NULL|88.0|NULL|8.34525|8.50646|9.21739|10.5814|0.0|NULL|9.53961|10.4053|12.1195|NULL|10.4|9.37735|NULL| NULL| NULL|NULL| NULL| NULL|NULL|NULL|NULL| 8.6|NULL|NULL|NULL| NULL| NULL|NULL|10.4| 9.8| 9.7|10.0|NULL|10.5|10.8|10.8|10.7|23000.0|0.0|   0.0|3.0|  2.0|23000.0|         []|     []|        []| []| []|  06201|1712836200000000000|\n",
      "|0-20000-0-06203|      P11-B|  52.36|3.3417| 41.84|NULL|225.6|218.0|NULL|4.0|237.7|7.63962| 9.07|0.35|NULL|NULL|8.73899|10.04|    0.0|    0.0| 0.0| 0.0|    0.0|    0.0| 0.0| 0.0| 0.0| 0.0| 0.0| 0.0| 0.0| 0.0| 0.0| 0.0|1025.29|1030.44|0.0|0.0|1027.5|0.0|NULL|NULL|NULL|NULL|NULL|1030.31|NULL|NULL|NULL|NULL|NULL|94.0|NULL|  7.926|8.26292|9.34106|12.9713|0.0|NULL|9.34829|10.6626|14.9103|NULL|11.4|10.8765|NULL| NULL| NULL|NULL| NULL| NULL|NULL|NULL|NULL|10.6|NULL|NULL|NULL| NULL| NULL|NULL|11.3|11.0|11.0|11.1|NULL|11.4|11.5|12.1|11.4|10800.0|0.0|   0.0|2.0|  2.0|10800.0|         []|     []|        []| []| []|  06203|1712836200000000000|\n",
      "|0-20000-0-06204|  K14-FA-1C|53.2694|3.6278|  41.8|NULL|249.5|243.3|NULL|3.3|257.3|8.39476|10.33|0.28|NULL|NULL|9.35519|11.07|27864.9|27864.9| 0.0| 0.0|27864.9|27864.9| 0.0| 0.0| 7.0| 7.0| 0.0| 0.0| 7.0| 7.0| 0.0| 0.0|1023.94|1029.08|0.0|0.0|1025.7|0.0|NULL|NULL|NULL|NULL|NULL|1028.97|NULL|NULL|NULL|NULL|NULL|91.0|NULL|8.73066|9.53248|10.4752|11.9948|0.0|NULL|10.7665| 11.882|13.6821|NULL|11.6|10.8055|NULL| NULL| NULL|NULL| NULL| NULL|NULL|NULL|NULL|10.3|NULL|NULL|NULL| NULL| NULL|NULL|11.6|10.9|10.9|11.0|NULL|11.7|12.0|12.0|12.0|22200.0|0.0|   0.0|3.0|  3.0|22200.0|         []|     []|        []| []| []|  06204|1712836200000000000|\n",
      "|0-20000-0-06205|    A12-CPP|55.3992|3.8103| 48.35|NULL|247.7|241.9|NULL|2.3|253.1|10.0239|12.02|0.35|NULL|NULL|11.2316|13.01|27434.9|27434.9| 0.0| 0.0|27434.9|27434.9| 0.0| 0.0| 2.0| 2.0| 0.0| 0.0| 2.0| 2.0| 0.0| 0.0|1018.12|1024.06|0.0|0.0|1019.9|0.0|NULL|NULL|NULL|NULL|NULL|1023.93|NULL|NULL|NULL|NULL|NULL|91.0|NULL|9.91552| 10.541|12.3423|14.0519|0.0|NULL|11.7669|14.4863| 18.069|NULL| 9.9|9.14861|NULL| NULL| NULL|NULL| NULL| NULL|NULL|NULL|NULL| 8.6|NULL|NULL|NULL| NULL| NULL|NULL| 9.8| 9.7| 9.5| 9.7|NULL| 9.9|11.2|11.2|11.2|16800.0|0.0|   0.0|1.0|  1.0|16800.0|         []|     []|        []| []| []|  06205|1712836200000000000|\n",
      "|0-20000-0-06207|    L9-FF-1|53.6144|4.9603| 45.31|NULL|229.3|225.0|NULL|3.2|236.3|8.61337|10.76|0.14|NULL|NULL|9.48118|11.36|   NULL|   NULL|NULL|NULL|   NULL|   NULL|NULL|NULL|NULL|NULL|NULL|NULL|NULL|NULL|NULL|NULL|1023.82|1029.39|0.0|0.0|1025.4|0.0|NULL|NULL|NULL|NULL|NULL|1029.27|NULL|NULL|NULL|NULL|NULL|92.0|NULL|8.71744|9.38986| 11.271|13.4164|0.0|NULL| 10.683|12.5108| 15.457|NULL|11.3|10.6019|NULL| NULL| NULL|NULL| NULL| NULL|NULL|NULL|NULL|10.0|NULL|NULL|NULL| NULL| NULL|NULL|11.3|10.6|10.6|10.9|NULL|11.3|11.5|11.5|11.5|11600.0|0.0|   0.0|0.0|  0.0|11600.0|         []|     []|        []| []| []|  06207|1712836200000000000|\n",
      "+---------------+-----------+-------+------+------+----+-----+-----+----+---+-----+-------+-----+----+----+----+-------+-----+-------+-------+----+----+-------+-------+----+----+----+----+----+----+----+----+----+----+-------+-------+---+---+------+---+----+----+----+----+----+-------+----+----+----+----+----+----+----+-------+-------+-------+-------+---+----+-------+-------+-------+----+----+-------+----+-----+-----+----+-----+-----+----+----+----+----+----+----+----+-----+-----+----+----+----+----+----+----+----+----+----+----+-------+---+------+---+-----+-------+-----------+-------+----------+---+---+-------+-------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "StructType([StructField('wsi', StringType(), True), StructField('stationname', StringType(), True), StructField('lat', DoubleType(), True), StructField('lon', DoubleType(), True), StructField('height', DoubleType(), True), StructField('D1H', DoubleType(), True), StructField('dd', DoubleType(), True), StructField('dn', DoubleType(), True), StructField('dr', DoubleType(), True), StructField('dsd', DoubleType(), True), StructField('dx', DoubleType(), True), StructField('ff', DoubleType(), True), StructField('ffs', DoubleType(), True), StructField('fsd', DoubleType(), True), StructField('fx', DoubleType(), True), StructField('fxs', DoubleType(), True), StructField('gff', DoubleType(), True), StructField('gffs', DoubleType(), True), StructField('h', DoubleType(), True), StructField('h1', DoubleType(), True), StructField('h2', DoubleType(), True), StructField('h3', DoubleType(), True), StructField('hc', DoubleType(), True), StructField('hc1', DoubleType(), True), StructField('hc2', DoubleType(), True), StructField('hc3', DoubleType(), True), StructField('n', DoubleType(), True), StructField('n1', DoubleType(), True), StructField('n2', DoubleType(), True), StructField('n3', DoubleType(), True), StructField('nc', DoubleType(), True), StructField('nc1', DoubleType(), True), StructField('nc2', DoubleType(), True), StructField('nc3', DoubleType(), True), StructField('p0', DoubleType(), True), StructField('pp', DoubleType(), True), StructField('pg', DoubleType(), True), StructField('pr', DoubleType(), True), StructField('ps', DoubleType(), True), StructField('pwc', DoubleType(), True), StructField('Q1H', DoubleType(), True), StructField('Q24H', DoubleType(), True), StructField('qg', DoubleType(), True), StructField('qgn', DoubleType(), True), StructField('qgx', DoubleType(), True), StructField('qnh', DoubleType(), True), StructField('R12H', DoubleType(), True), StructField('R1H', DoubleType(), True), StructField('R24H', DoubleType(), True), StructField('R6H', DoubleType(), True), StructField('rg', DoubleType(), True), StructField('rh', DoubleType(), True), StructField('rh10', DoubleType(), True), StructField('Sav1H', DoubleType(), True), StructField('Sax1H', DoubleType(), True), StructField('Sax3H', DoubleType(), True), StructField('Sax6H', DoubleType(), True), StructField('sq', DoubleType(), True), StructField('ss', DoubleType(), True), StructField('Sx1H', DoubleType(), True), StructField('Sx3H', DoubleType(), True), StructField('Sx6H', DoubleType(), True), StructField('t10', DoubleType(), True), StructField('ta', DoubleType(), True), StructField('tb', DoubleType(), True), StructField('tb1', DoubleType(), True), StructField('Tb1n6', DoubleType(), True), StructField('Tb1x6', DoubleType(), True), StructField('tb2', DoubleType(), True), StructField('Tb2n6', DoubleType(), True), StructField('Tb2x6', DoubleType(), True), StructField('tb3', DoubleType(), True), StructField('tb4', DoubleType(), True), StructField('tb5', DoubleType(), True), StructField('td', DoubleType(), True), StructField('td10', DoubleType(), True), StructField('tg', DoubleType(), True), StructField('tgn', DoubleType(), True), StructField('Tgn12', DoubleType(), True), StructField('Tgn14', DoubleType(), True), StructField('Tgn6', DoubleType(), True), StructField('tn', DoubleType(), True), StructField('Tn12', DoubleType(), True), StructField('Tn14', DoubleType(), True), StructField('Tn6', DoubleType(), True), StructField('tsd', DoubleType(), True), StructField('tx', DoubleType(), True), StructField('Tx12', DoubleType(), True), StructField('Tx24', DoubleType(), True), StructField('Tx6', DoubleType(), True), StructField('vv', DoubleType(), True), StructField('W10', DoubleType(), True), StructField('W10-10', DoubleType(), True), StructField('ww', DoubleType(), True), StructField('ww-10', DoubleType(), True), StructField('zm', DoubleType(), True), StructField('iso_dataset', BinaryType(), True), StructField('product', BinaryType(), True), StructField('projection', BinaryType(), True), StructField('nhc', BinaryType(), True), StructField('za', BinaryType(), True), StructField('station', StringType(), True), StructField('time', LongType(), True)])"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "StructType([StructField('wsi', StringType(), True), StructField('stationname', StringType(), True), StructField('lat', DoubleType(), True), StructField('lon', DoubleType(), True), StructField('height', DoubleType(), True), StructField('D1H', DoubleType(), True), StructField('dd', DoubleType(), True), StructField('dn', DoubleType(), True), StructField('dr', DoubleType(), True), StructField('dsd', DoubleType(), True), StructField('dx', DoubleType(), True), StructField('ff', DoubleType(), True), StructField('ffs', DoubleType(), True), StructField('fsd', DoubleType(), True), StructField('fx', DoubleType(), True), StructField('fxs', DoubleType(), True), StructField('gff', DoubleType(), True), StructField('gffs', DoubleType(), True), StructField('h', DoubleType(), True), StructField('h1', DoubleType(), True), StructField('h2', DoubleType(), True), StructField('h3', DoubleType(), True), StructField('hc', DoubleType(), True), StructField('hc1', DoubleType(), True), StructField('hc2', DoubleType(), True), StructField('hc3', DoubleType(), True), StructField('n', DoubleType(), True), StructField('n1', DoubleType(), True), StructField('n2', DoubleType(), True), StructField('n3', DoubleType(), True), StructField('nc', DoubleType(), True), StructField('nc1', DoubleType(), True), StructField('nc2', DoubleType(), True), StructField('nc3', DoubleType(), True), StructField('p0', DoubleType(), True), StructField('pp', DoubleType(), True), StructField('pg', DoubleType(), True), StructField('pr', DoubleType(), True), StructField('ps', DoubleType(), True), StructField('pwc', DoubleType(), True), StructField('Q1H', DoubleType(), True), StructField('Q24H', DoubleType(), True), StructField('qg', DoubleType(), True), StructField('qgn', DoubleType(), True), StructField('qgx', DoubleType(), True), StructField('qnh', DoubleType(), True), StructField('R12H', DoubleType(), True), StructField('R1H', DoubleType(), True), StructField('R24H', DoubleType(), True), StructField('R6H', DoubleType(), True), StructField('rg', DoubleType(), True), StructField('rh', DoubleType(), True), StructField('rh10', DoubleType(), True), StructField('Sav1H', DoubleType(), True), StructField('Sax1H', DoubleType(), True), StructField('Sax3H', DoubleType(), True), StructField('Sax6H', DoubleType(), True), StructField('sq', DoubleType(), True), StructField('ss', DoubleType(), True), StructField('Sx1H', DoubleType(), True), StructField('Sx3H', DoubleType(), True), StructField('Sx6H', DoubleType(), True), StructField('t10', DoubleType(), True), StructField('ta', DoubleType(), True), StructField('tb', DoubleType(), True), StructField('tb1', DoubleType(), True), StructField('Tb1n6', DoubleType(), True), StructField('Tb1x6', DoubleType(), True), StructField('tb2', DoubleType(), True), StructField('Tb2n6', DoubleType(), True), StructField('Tb2x6', DoubleType(), True), StructField('tb3', DoubleType(), True), StructField('tb4', DoubleType(), True), StructField('tb5', DoubleType(), True), StructField('td', DoubleType(), True), StructField('td10', DoubleType(), True), StructField('tg', DoubleType(), True), StructField('tgn', DoubleType(), True), StructField('Tgn12', DoubleType(), True), StructField('Tgn14', DoubleType(), True), StructField('Tgn6', DoubleType(), True), StructField('tn', DoubleType(), True), StructField('Tn12', DoubleType(), True), StructField('Tn14', DoubleType(), True), StructField('Tn6', DoubleType(), True), StructField('tsd', DoubleType(), True), StructField('tx', DoubleType(), True), StructField('Tx12', DoubleType(), True), StructField('Tx24', DoubleType(), True), StructField('Tx6', DoubleType(), True), StructField('vv', DoubleType(), True), StructField('W10', DoubleType(), True), StructField('W10-10', DoubleType(), True), StructField('ww', DoubleType(), True), StructField('ww-10', DoubleType(), True), StructField('zm', DoubleType(), True), StructField('iso_dataset', BinaryType(), True), StructField('product', BinaryType(), True), StructField('projection', BinaryType(), True), StructField('nhc', BinaryType(), True), StructField('za', BinaryType(), True), StructField('station', StringType(), True), StructField('time', LongType(), True)])"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.sql.types import StructType, StructField, StringType, DoubleType, LongType, BinaryType\n",
    "\n",
    "StructType(\n",
    "    [\n",
    "        StructField(\"wsi\", StringType(), True),\n",
    "        StructField(\"stationname\", StringType(), True),\n",
    "        StructField(\"lat\", DoubleType(), True),\n",
    "        StructField(\"lon\", DoubleType(), True),\n",
    "        StructField(\"height\", DoubleType(), True),\n",
    "        StructField(\"D1H\", DoubleType(), True),\n",
    "        StructField(\"dd\", DoubleType(), True),\n",
    "        StructField(\"dn\", DoubleType(), True),\n",
    "        StructField(\"dr\", DoubleType(), True),\n",
    "        StructField(\"dsd\", DoubleType(), True),\n",
    "        StructField(\"dx\", DoubleType(), True),\n",
    "        StructField(\"ff\", DoubleType(), True),\n",
    "        StructField(\"ffs\", DoubleType(), True),\n",
    "        StructField(\"fsd\", DoubleType(), True),\n",
    "        StructField(\"fx\", DoubleType(), True),\n",
    "        StructField(\"fxs\", DoubleType(), True),\n",
    "        StructField(\"gff\", DoubleType(), True),\n",
    "        StructField(\"gffs\", DoubleType(), True),\n",
    "        StructField(\"h\", DoubleType(), True),\n",
    "        StructField(\"h1\", DoubleType(), True),\n",
    "        StructField(\"h2\", DoubleType(), True),\n",
    "        StructField(\"h3\", DoubleType(), True),\n",
    "        StructField(\"hc\", DoubleType(), True),\n",
    "        StructField(\"hc1\", DoubleType(), True),\n",
    "        StructField(\"hc2\", DoubleType(), True),\n",
    "        StructField(\"hc3\", DoubleType(), True),\n",
    "        StructField(\"n\", DoubleType(), True),\n",
    "        StructField(\"n1\", DoubleType(), True),\n",
    "        StructField(\"n2\", DoubleType(), True),\n",
    "        StructField(\"n3\", DoubleType(), True),\n",
    "        StructField(\"nc\", DoubleType(), True),\n",
    "        StructField(\"nc1\", DoubleType(), True),\n",
    "        StructField(\"nc2\", DoubleType(), True),\n",
    "        StructField(\"nc3\", DoubleType(), True),\n",
    "        StructField(\"p0\", DoubleType(), True),\n",
    "        StructField(\"pp\", DoubleType(), True),\n",
    "        StructField(\"pg\", DoubleType(), True),\n",
    "        StructField(\"pr\", DoubleType(), True),\n",
    "        StructField(\"ps\", DoubleType(), True),\n",
    "        StructField(\"pwc\", DoubleType(), True),\n",
    "        StructField(\"Q1H\", DoubleType(), True),\n",
    "        StructField(\"Q24H\", DoubleType(), True),\n",
    "        StructField(\"qg\", DoubleType(), True),\n",
    "        StructField(\"qgn\", DoubleType(), True),\n",
    "        StructField(\"qgx\", DoubleType(), True),\n",
    "        StructField(\"qnh\", DoubleType(), True),\n",
    "        StructField(\"R12H\", DoubleType(), True),\n",
    "        StructField(\"R1H\", DoubleType(), True),\n",
    "        StructField(\"R24H\", DoubleType(), True),\n",
    "        StructField(\"R6H\", DoubleType(), True),\n",
    "        StructField(\"rg\", DoubleType(), True),\n",
    "        StructField(\"rh\", DoubleType(), True),\n",
    "        StructField(\"rh10\", DoubleType(), True),\n",
    "        StructField(\"Sav1H\", DoubleType(), True),\n",
    "        StructField(\"Sax1H\", DoubleType(), True),\n",
    "        StructField(\"Sax3H\", DoubleType(), True),\n",
    "        StructField(\"Sax6H\", DoubleType(), True),\n",
    "        StructField(\"sq\", DoubleType(), True),\n",
    "        StructField(\"ss\", DoubleType(), True),\n",
    "        StructField(\"Sx1H\", DoubleType(), True),\n",
    "        StructField(\"Sx3H\", DoubleType(), True),\n",
    "        StructField(\"Sx6H\", DoubleType(), True),\n",
    "        StructField(\"t10\", DoubleType(), True),\n",
    "        StructField(\"ta\", DoubleType(), True),\n",
    "        StructField(\"tb\", DoubleType(), True),\n",
    "        StructField(\"tb1\", DoubleType(), True),\n",
    "        StructField(\"Tb1n6\", DoubleType(), True),\n",
    "        StructField(\"Tb1x6\", DoubleType(), True),\n",
    "        StructField(\"tb2\", DoubleType(), True),\n",
    "        StructField(\"Tb2n6\", DoubleType(), True),\n",
    "        StructField(\"Tb2x6\", DoubleType(), True),\n",
    "        StructField(\"tb3\", DoubleType(), True),\n",
    "        StructField(\"tb4\", DoubleType(), True),\n",
    "        StructField(\"tb5\", DoubleType(), True),\n",
    "        StructField(\"td\", DoubleType(), True),\n",
    "        StructField(\"td10\", DoubleType(), True),\n",
    "        StructField(\"tg\", DoubleType(), True),\n",
    "        StructField(\"tgn\", DoubleType(), True),\n",
    "        StructField(\"Tgn12\", DoubleType(), True),\n",
    "        StructField(\"Tgn14\", DoubleType(), True),\n",
    "        StructField(\"Tgn6\", DoubleType(), True),\n",
    "        StructField(\"tn\", DoubleType(), True),\n",
    "        StructField(\"Tn12\", DoubleType(), True),\n",
    "        StructField(\"Tn14\", DoubleType(), True),\n",
    "        StructField(\"Tn6\", DoubleType(), True),\n",
    "        StructField(\"tsd\", DoubleType(), True),\n",
    "        StructField(\"tx\", DoubleType(), True),\n",
    "        StructField(\"Tx12\", DoubleType(), True),\n",
    "        StructField(\"Tx24\", DoubleType(), True),\n",
    "        StructField(\"Tx6\", DoubleType(), True),\n",
    "        StructField(\"vv\", DoubleType(), True),\n",
    "        StructField(\"W10\", DoubleType(), True),\n",
    "        StructField(\"W10-10\", DoubleType(), True),\n",
    "        StructField(\"ww\", DoubleType(), True),\n",
    "        StructField(\"ww-10\", DoubleType(), True),\n",
    "        StructField(\"zm\", DoubleType(), True),\n",
    "        StructField(\"iso_dataset\", BinaryType(), True),\n",
    "        StructField(\"product\", BinaryType(), True),\n",
    "        StructField(\"projection\", BinaryType(), True),\n",
    "        StructField(\"nhc\", BinaryType(), True),\n",
    "        StructField(\"za\", BinaryType(), True),\n",
    "        StructField(\"station\", StringType(), True),\n",
    "        StructField(\"time\", LongType(), True),\n",
    "    ]\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 5:>                                                          (0 + 5) / 5]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-------+-----------+\n",
      "|         stationname|station|iso_dataset|\n",
      "+--------------------+-------+-----------+\n",
      "|        BERKHOUT AWS|  06249|      false|\n",
      "|               K13-A|  06252|      false|\n",
      "|            D15-FA-1|  06201|      false|\n",
      "|       WESTDORPE AWS|  06319|      false|\n",
      "|           IJMOND WP|  06209|      false|\n",
      "|    WIJK AAN ZEE AWS|  06257|      false|\n",
      "|         GILZE RIJEN|  06350|      false|\n",
      "|AMSTERDAM/SCHIPHO...|  06240|      false|\n",
      "|          Muiden BTP|  06236|      false|\n",
      "|       EURO PLATFORM|  06321|      false|\n",
      "|       MARKNESSE AWS|  06273|      false|\n",
      "|       Nieuwkoop BTP|  06238|      false|\n",
      "|         IJMUIDEN WP|  06225|      false|\n",
      "|           K14-FA-1C|  06204|      false|\n",
      "| BORSSELE ALFA (BSA)|  06317|      false|\n",
      "|ROTTERDAM THE HAG...|  06344|      false|\n",
      "|               AWG-1|  06208|      false|\n",
      "|              DEELEN|  06275|      false|\n",
      "|TERSCHELLING HOOR...|  06251|      false|\n",
      "|          STAVENISSE|  06324|      false|\n",
      "+--------------------+-------+-----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "df.select(df.stationname, df.station, (df.iso_dataset.isNull()).alias(\"iso_dataset\")).distinct().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Save to Big Query"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "24/04/12 16:55:53 ERROR StandaloneSchedulerBackend: Application has been killed. Reason: Master removed our application: KILLED\n",
      "24/04/12 16:55:55 ERROR Inbox: Ignoring error\n",
      "org.apache.spark.SparkException: Exiting due to error from cluster scheduler: Master removed our application: KILLED\n",
      "\tat org.apache.spark.errors.SparkCoreErrors$.clusterSchedulerError(SparkCoreErrors.scala:291)\n",
      "\tat org.apache.spark.scheduler.TaskSchedulerImpl.error(TaskSchedulerImpl.scala:981)\n",
      "\tat org.apache.spark.scheduler.cluster.StandaloneSchedulerBackend.dead(StandaloneSchedulerBackend.scala:165)\n",
      "\tat org.apache.spark.deploy.client.StandaloneAppClient$ClientEndpoint.markDead(StandaloneAppClient.scala:263)\n",
      "\tat org.apache.spark.deploy.client.StandaloneAppClient$ClientEndpoint$$anonfun$receive$1.applyOrElse(StandaloneAppClient.scala:170)\n",
      "\tat org.apache.spark.rpc.netty.Inbox.$anonfun$process$1(Inbox.scala:115)\n",
      "\tat org.apache.spark.rpc.netty.Inbox.safelyCall(Inbox.scala:213)\n",
      "\tat org.apache.spark.rpc.netty.Inbox.process(Inbox.scala:100)\n",
      "\tat org.apache.spark.rpc.netty.MessageLoop.org$apache$spark$rpc$netty$MessageLoop$$receiveLoop(MessageLoop.scala:75)\n",
      "\tat org.apache.spark.rpc.netty.MessageLoop$$anon$1.run(MessageLoop.scala:41)\n",
      "\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)\n",
      "\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)\n",
      "\tat java.base/java.lang.Thread.run(Unknown Source)\n"
     ]
    }
   ],
   "source": [
    "(\n",
    "    df.write.format(\"bigquery\")\n",
    "    .option(\"temporaryGcsBucket\", \"weather_data_de_bucket\")\n",
    "    .mode(\"append\")\n",
    "    .option(\"parentProject\", \"dataengineeringbootcamp-419022\")\n",
    "    .save(\"dataengineeringbootcamp-419022:weather_data_de.weather_data\")\n",
    ")"
   ]
  }
 ],
 "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.11.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}