examples/basic_write_from_pandas.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [],
   "source": [
    "from datetime import date\n",
    "import pandas as pd\n",
    "\n",
    "from py2k.record import PandasToRecordsTransformer"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Creating a pandas dataframe for example purpose:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "output_type": "execute_result",
     "data": {
      "text/plain": [
       "     name cool_level   value        date\n",
       "0  Daniel        low    27.1  2021-03-01\n",
       "1   David       high   100.0  2021-03-01\n",
       "2  Felipe       high     0.0  2021-03-01\n",
       "3  Ruslan       high  9000.0  2021-03-01"
      ],
      "text/html": "<div>\n<style scoped>\n    .dataframe tbody tr th:only-of-type {\n        vertical-align: middle;\n    }\n\n    .dataframe tbody tr th {\n        vertical-align: top;\n    }\n\n    .dataframe thead th {\n        text-align: right;\n    }\n</style>\n<table border=\"1\" class=\"dataframe\">\n  <thead>\n    <tr style=\"text-align: right;\">\n      <th></th>\n      <th>name</th>\n      <th>cool_level</th>\n      <th>value</th>\n      <th>date</th>\n    </tr>\n  </thead>\n  <tbody>\n    <tr>\n      <th>0</th>\n      <td>Daniel</td>\n      <td>low</td>\n      <td>27.1</td>\n      <td>2021-03-01</td>\n    </tr>\n    <tr>\n      <th>1</th>\n      <td>David</td>\n      <td>high</td>\n      <td>100.0</td>\n      <td>2021-03-01</td>\n    </tr>\n    <tr>\n      <th>2</th>\n      <td>Felipe</td>\n      <td>high</td>\n      <td>0.0</td>\n      <td>2021-03-01</td>\n    </tr>\n    <tr>\n      <th>3</th>\n      <td>Ruslan</td>\n      <td>high</td>\n      <td>9000.0</td>\n      <td>2021-03-01</td>\n    </tr>\n  </tbody>\n</table>\n</div>"
     },
     "metadata": {},
     "execution_count": 45
    }
   ],
   "source": [
    "df = pd.DataFrame({'name': ['Daniel','David','Felipe','Ruslan'],\n",
    "                   'cool_level':['low', 'high', 'high', 'high'],\n",
    "                   'value': [27.1,100.0, 0, 9000.0],\n",
    "                   'date': [date(2021,3,1),date(2021,3,1),date(2021,3,1),date(2021,3,1)]})\n",
    "df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We create a basic `PandasToRecordsTransformer` by passing the dataframe and the `record_name`. The `record_name` auto assigns the \"name\" and \"namespace\" within the generated schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [
    {
     "output_type": "execute_result",
     "data": {
      "text/plain": [
       "name           object\n",
       "cool_level     object\n",
       "value         float64\n",
       "date           object\n",
       "dtype: object"
      ]
     },
     "metadata": {},
     "execution_count": 46
    }
   ],
   "source": [
    "df.dtypes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [],
   "source": [
    "record_transformer = PandasToRecordsTransformer(df, record_name='KafkaRecord')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [],
   "source": [
    "records = record_transformer.from_pandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "output_type": "execute_result",
     "data": {
      "text/plain": [
       "[KafkaRecord(name='Daniel', cool_level='low', value=27.1, date=datetime.date(2021, 3, 1)),\n",
       " KafkaRecord(name='David', cool_level='high', value=100.0, date=datetime.date(2021, 3, 1)),\n",
       " KafkaRecord(name='Felipe', cool_level='high', value=0.0, date=datetime.date(2021, 3, 1)),\n",
       " KafkaRecord(name='Ruslan', cool_level='high', value=9000.0, date=datetime.date(2021, 3, 1))]"
      ]
     },
     "metadata": {},
     "execution_count": 49
    }
   ],
   "source": [
    "records"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [
    {
     "output_type": "execute_result",
     "data": {
      "text/plain": [
       "{'type': 'record',\n",
       " 'name': 'KafkaRecord',\n",
       " 'namespace': 'python.kafka.kafkarecord',\n",
       " 'fields': [{'default': '', 'type': 'string', 'name': 'name'},\n",
       "  {'default': '', 'type': 'string', 'name': 'cool_level'},\n",
       "  {'default': -1.0, 'type': 'double', 'name': 'value'},\n",
       "  {'default': '0001-01-01',\n",
       "   'type': 'string',\n",
       "   'format': 'date',\n",
       "   'name': 'date'}]}"
      ]
     },
     "metadata": {},
     "execution_count": 50
    }
   ],
   "source": [
    "# We can inspect the schema of a record as follows:\n",
    "records[0].schema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To write these records onto Kafka, we need to create a `KafkaWriter` object. To this we need to specify a `producer_config` and `schema_registry_config`. The configuration for these can be found in the documentation for the [python confluent kafka](https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html) package."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [],
   "source": [
    "from py2k.writer import KafkaWriter"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [],
   "source": [
    "topic = 'py2k-test-topic'\n",
    "\n",
    "writer = KafkaWriter(topic=topic,\n",
    "                     schema_registry_config=schema_registry_config,\n",
    "                     producer_config=producer_config)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [
    {
     "output_type": "stream",
     "name": "stderr",
     "text": [
      "100%|██████████| 4/4 [00:00<00:00,  7.69it/s]\n"
     ]
    }
   ],
   "source": [
    "writer.write(records)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}