kafka_reassign_cluster_partitions.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from kafka import KafkaProducer\n",
    "import json\n",
    "import random\n",
    "import json "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Datacenter's name: list of the brokers\n",
    "hosts_dict = {0:[0, 1, 2], 1: [3, 4, 5], 2: [6, 7, 8]}\n",
    "\n",
    "topics_list = [\"__consumer_offsets\", \"partitioned-topic\", \"fast-messages\n",
    "\", \"t1\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[0, 1, 2]"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# List of the datacenters\n",
    "hosts_dict[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "{\n",
      "    \"partitions\": [{\n",
      "        \"topic\": \"foo\",\n",
      "        \"partition\": 1,\n",
      "        \"replicas\": [1, 2, 4]\n",
      "    }],\n",
      "    \"version\": 1\n",
      "}\n",
      "\n"
     ]
    }
   ],
   "source": [
    "default_json = \"\"\"\n",
    "{\n",
    "    \"partitions\": [{\n",
    "        \"topic\": \"foo\",\n",
    "        \"partition\": 1,\n",
    "        \"replicas\": [1, 2, 4]\n",
    "    }],\n",
    "    \"version\": 1\n",
    "}\n",
    "\"\"\"\n",
    "print(default_json)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{'version': 1, 'partitions': []}\n",
      "{'partition': 1, 'topic': 'foo', 'replicas': [1, 2, 4]}\n"
     ]
    }
   ],
   "source": [
    "parsed_string = json.loads(default_json)\n",
    "desc_of_each_part = parsed_string['partitions'][0].copy()\n",
    "parsed_string['partitions'].clear()\n",
    "\n",
    "print(parsed_string)\n",
    "print(desc_of_each_part)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def get_partitions_list(topic):\n",
    "    producer = KafkaProducer(bootstrap_servers='localhost:9092')\n",
    "    partitions_list = producer.partitions_for(topic)\n",
    "    return partitions_list"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Generate new list of the brokers (for storage partitions for each topics on the new datacenters)\n",
    "def create_extended_broker_list():\n",
    "    updated_broker_list = []\n",
    "    \n",
    "    for datacenter_num in hosts_dict:\n",
    "        # List of the brokers in the current datacenter\n",
    "        broker_list = hosts_dict[datacenter_num]\n",
    "        current_broker = random.choice(broker_list)\n",
    "        updated_broker_list.append(current_broker)\n",
    "\n",
    "    return updated_broker_list"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def create_json_for_partitions():\n",
    "    \n",
    "    # For each topic:\n",
    "    for topic in topics_list:\n",
    "        partitions_list = get_partitions_list(topic) \n",
    "    \n",
    "        for partitions_num in partitions_list:\n",
    "            desc_of_each_part['partition'] = partitions_num\n",
    "            # Write the name of the topic without \\n\n",
    "            desc_of_each_part['topic'] = topic\n",
    "            # Determine brokers for storage partitions for this topic\n",
    "            desc_of_each_part['replicas'] = create_extended_broker_list()\n",
    "            # Add new info to json\n",
    "            parsed_string['partitions'].append(desc_of_each_part.copy())\n",
    "    return parsed_string"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{\n",
      "    \"partitions\": [\n",
      "        {\n",
      "            \"partition\": 0,\n",
      "            \"replicas\": [\n",
      "                0,\n",
      "                5,\n",
      "                8\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 1,\n",
      "            \"replicas\": [\n",
      "                0,\n",
      "                3,\n",
      "                8\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 2,\n",
      "            \"replicas\": [\n",
      "                0,\n",
      "                3,\n",
      "                6\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 3,\n",
      "            \"replicas\": [\n",
      "                0,\n",
      "                3,\n",
      "                6\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 4,\n",
      "            \"replicas\": [\n",
      "                1,\n",
      "                5,\n",
      "                8\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 5,\n",
      "            \"replicas\": [\n",
      "                0,\n",
      "                5,\n",
      "                7\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 6,\n",
      "            \"replicas\": [\n",
      "                1,\n",
      "                3,\n",
      "                8\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 7,\n",
      "            \"replicas\": [\n",
      "                0,\n",
      "                3,\n",
      "                8\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 8,\n",
      "            \"replicas\": [\n",
      "                0,\n",
      "                3,\n",
      "                8\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 9,\n",
      "            \"replicas\": [\n",
      "                1,\n",
      "                3,\n",
      "                7\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 10,\n",
      "            \"replicas\": [\n",
      "                1,\n",
      "                3,\n",
      "                7\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 11,\n",
      "            \"replicas\": [\n",
      "                1,\n",
      "                4,\n",
      "                8\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 12,\n",
      "            \"replicas\": [\n",
      "                0,\n",
      "                4,\n",
      "                6\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 13,\n",
      "            \"replicas\": [\n",
      "                0,\n",
      "                3,\n",
      "                8\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 14,\n",
      "            \"replicas\": [\n",
      "                0,\n",
      "                5,\n",
      "                7\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 15,\n",
      "            \"replicas\": [\n",
      "                0,\n",
      "                4,\n",
      "                6\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 16,\n",
      "            \"replicas\": [\n",
      "                1,\n",
      "                3,\n",
      "                7\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 17,\n",
      "            \"replicas\": [\n",
      "                2,\n",
      "                5,\n",
      "                7\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 18,\n",
      "            \"replicas\": [\n",
      "                2,\n",
      "                3,\n",
      "                6\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 19,\n",
      "            \"replicas\": [\n",
      "                1,\n",
      "                4,\n",
      "                6\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 20,\n",
      "            \"replicas\": [\n",
      "                0,\n",
      "                4,\n",
      "                7\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 21,\n",
      "            \"replicas\": [\n",
      "                1,\n",
      "                5,\n",
      "                6\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 22,\n",
      "            \"replicas\": [\n",
      "                2,\n",
      "                5,\n",
      "                6\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 23,\n",
      "            \"replicas\": [\n",
      "                1,\n",
      "                5,\n",
      "                6\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 24,\n",
      "            \"replicas\": [\n",
      "                0,\n",
      "                4,\n",
      "                8\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 25,\n",
      "            \"replicas\": [\n",
      "                0,\n",
      "                3,\n",
      "                6\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 26,\n",
      "            \"replicas\": [\n",
      "                1,\n",
      "                4,\n",
      "                6\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 27,\n",
      "            \"replicas\": [\n",
      "                2,\n",
      "                3,\n",
      "                7\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 28,\n",
      "            \"replicas\": [\n",
      "                0,\n",
      "                3,\n",
      "                6\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 29,\n",
      "            \"replicas\": [\n",
      "                2,\n",
      "                5,\n",
      "                7\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 30,\n",
      "            \"replicas\": [\n",
      "                2,\n",
      "                5,\n",
      "                8\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 31,\n",
      "            \"replicas\": [\n",
      "                0,\n",
      "                4,\n",
      "                8\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 32,\n",
      "            \"replicas\": [\n",
      "                0,\n",
      "                5,\n",
      "                8\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 33,\n",
      "            \"replicas\": [\n",
      "                1,\n",
      "                5,\n",
      "                8\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 34,\n",
      "            \"replicas\": [\n",
      "                0,\n",
      "                3,\n",
      "                8\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 35,\n",
      "            \"replicas\": [\n",
      "                1,\n",
      "                4,\n",
      "                8\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 36,\n",
      "            \"replicas\": [\n",
      "                2,\n",
      "                3,\n",
      "                7\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 37,\n",
      "            \"replicas\": [\n",
      "                0,\n",
      "                3,\n",
      "                7\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 38,\n",
      "            \"replicas\": [\n",
      "                1,\n",
      "                4,\n",
      "                8\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 39,\n",
      "            \"replicas\": [\n",
      "                1,\n",
      "                5,\n",
      "                7\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 40,\n",
      "            \"replicas\": [\n",
      "                2,\n",
      "                3,\n",
      "                8\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 41,\n",
      "            \"replicas\": [\n",
      "                2,\n",
      "                4,\n",
      "                6\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 42,\n",
      "            \"replicas\": [\n",
      "                0,\n",
      "                5,\n",
      "                7\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 43,\n",
      "            \"replicas\": [\n",
      "                1,\n",
      "                4,\n",
      "                8\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 44,\n",
      "            \"replicas\": [\n",
      "                0,\n",
      "                3,\n",
      "                6\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 45,\n",
      "            \"replicas\": [\n",
      "                0,\n",
      "                4,\n",
      "                6\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 46,\n",
      "            \"replicas\": [\n",
      "                2,\n",
      "                5,\n",
      "                7\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 47,\n",
      "            \"replicas\": [\n",
      "                1,\n",
      "                3,\n",
      "                7\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 48,\n",
      "            \"replicas\": [\n",
      "                0,\n",
      "                4,\n",
      "                7\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 49,\n",
      "            \"replicas\": [\n",
      "                0,\n",
      "                3,\n",
      "                8\n",
      "            ],\n",
      "            \"topic\": \"__consumer_offsets\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 0,\n",
      "            \"replicas\": [\n",
      "                0,\n",
      "                5,\n",
      "                6\n",
      "            ],\n",
      "            \"topic\": \"partitioned-topic\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 1,\n",
      "            \"replicas\": [\n",
      "                1,\n",
      "                4,\n",
      "                8\n",
      "            ],\n",
      "            \"topic\": \"partitioned-topic\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 2,\n",
      "            \"replicas\": [\n",
      "                1,\n",
      "                4,\n",
      "                7\n",
      "            ],\n",
      "            \"topic\": \"partitioned-topic\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 0,\n",
      "            \"replicas\": [\n",
      "                2,\n",
      "                5,\n",
      "                7\n",
      "            ],\n",
      "            \"topic\": \"t0\"\n",
      "        },\n",
      "        {\n",
      "            \"partition\": 0,\n",
      "            \"replicas\": [\n",
      "                0,\n",
      "                4,\n",
      "                7\n",
      "            ],\n",
      "            \"topic\": \"t1\"\n",
      "        }\n",
      "    ],\n",
      "    \"version\": 1\n",
      "}\n"
     ]
    }
   ],
   "source": [
    "def main():\n",
    "    list_for_json = create_json_for_partitions()\n",
    "    result_json = json.dumps(list_for_json)\n",
    "    \n",
    "    # Magic happens here to make it pretty-printed\n",
    "    pretty_json = json.dumps(json.loads(result_json), indent=4, sort_keys=True)\n",
    "    print(pretty_json)\n",
    "        \n",
    "    # Now write output to a file\n",
    "    json_file = open(\"partitions-to-move.json\", \"w\")\n",
    "    json_file.write(pretty_json)\n",
    "    json_file.close()  \n",
    "\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    main()"
   ]
  }
 ],
 "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.5.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}