{"id":15690,"date":"2022-02-18T14:53:01","date_gmt":"2022-02-18T13:53:01","guid":{"rendered":"https:\/\/scriptfactory.pl\/blog\/?p=15690"},"modified":"2022-02-18T14:56:11","modified_gmt":"2022-02-18T13:56:11","slug":"migracja-zasobow-azure-databricks-cz-1","status":"publish","type":"post","link":"https:\/\/scriptfactory.pl\/blog\/azure\/databricks\/migracja-zasobow-azure-databricks-cz-1\/","title":{"rendered":"Migracja zasob\u00f3w Azure Databricks cz. 1"},"content":{"rendered":"\n

Wprowadzenie:<\/h2>\n\n\n\n

Azure Databricks jest doskona\u0142\u0105 platform\u0105 do wykonania analiz , czy procesowania znacznej ilo\u015bci danych tzw. Big Data<\/a>. Jednak nie jest to jeszcze do ko\u0144ca dojrza\u0142a platforma. Przez co cz\u0119sto pojawiaj\u0105 si\u0119 problemy zwi\u0105zane np. z procesem migracji danych\/klastr\u00f3w\/job\u00f3w\/kluczy\/notebook\u00f3w itp. do innych \u015brodowisk (takich jak Dev\/Test\/QA). Aby tego dokona\u0107 ka\u017cdy ze wspomnianych komponent\u00f3w migruje si\u0119 w inny spos\u00f3b. Jest jednak narz\u0119dzie przygotowane przez Databricks Labs<\/a>, kt\u00f3re u\u0142atwia ten proces.<\/p>\n\n\n\n

Pami\u0119tajmy \u017ce Databricks jest platform\u0105 skupion\u0105 na podej\u015bciu API First, czyli \u017ce najpierw implementowany jest kod u\u017cywany w API, a nast\u0119pnie dobudowywane s\u0105 elementy takie jak GUI , klasy w python\/cli czy innych j\u0119zykach. Dzi\u0119ki temu mamy sami mo\u017cemy zbudowa\u0107\/rozbudowa\u0107 narz\u0119dzia migracyjne poszczeg\u00f3lnych komponent\u00f3w znaj\u0105c tylko definicj\u0119 API.<\/p>\n\n\n\n

\"\"
lista API<\/figcaption><\/figure>\n\n\n\n

Github:<\/strong><\/h2>\n\n\n\n

Kod migracji platformy Databricks<\/a><\/p>\n\n\n\n

Na stronie w readme znajduje si\u0119 do\u015b\u0107 rozbudowana dokumentacja jak u\u017cywa\u0107 kodu Python. My jednak dodamy co\u015b od siebie, tzn. opakujemy metody export\/import w jedn\u0105 klase do migracji konkretnych zasob\u00f3w.<\/p>\n\n\n\n

Implementacja w Azure Databricks:<\/h2>\n\n\n\n

G\u0142\u00f3wne pliki wywo\u0142ywane w oryginalnej implementacji to export_db.py oraz import_db.py. Czyli operacje export\/import s\u0105 niezale\u017cne od siebie, co za tym idzie aby wykona\u0107 migracj\u0119 pomi\u0119dzy \u015brodowiskami, nale\u017cy wykona\u0107 seri\u0119 skrypt\u00f3w aby dane najpierw wyeksportowa\u0107 a nast\u0119pnie zaimportowa\u0107. My przygotowali\u015bmy migrate_db.py kt\u00f3ry wykonuje odpowiednie operacje w kolejno\u015bci, aby m\u00f3c z migrowa\u0107 ca\u0142y workspace. Opis tego procesu jak r\u00f3wnie\u017c dok\u0142adne kody implementacji rozbijemy na kilka artyku\u0142\u00f3w. W pierwszym z nich skupimy si\u0119 na u\u017cytkownikach i klastrach.<\/p>\n\n\n\n

Zaczynamy od definicji migrate_db.py<\/p>\n\n\n\n

from dbclient import *\nfrom timeit import default_timer as timer\nfrom datetime import timedelta, datetime\nfrom os import makedirs\n\n\n# python 3.6\ndef main():\n    my_parser = get_migrate_parser()\n    args = my_parser.parse_args()\n\n    import_login_args = get_login_credentials(profile=args.import_profile)\n    export_login_args = get_login_credentials(profile=args.export_profile)\n\n    if is_azure_creds(import_login_args) and (not args.azure):\n        raise ValueError('Login credentials do not match args. Please provide --azure flag for azure envs.')\n\n    if is_azure_creds(export_login_args) and (not args.azure):\n        raise ValueError('Login credentials do not match args. Please provide --azure flag for azure envs.')\n    import_url = import_login_args['host']\n    import_token = import_login_args['token']\n    export_url = export_login_args['host']\n    export_token = export_login_args['token']\n    \n    import_client_config = build_client_config(import_url, import_token, args)\n    export_client_config = build_client_config(export_url, export_token, args)\n\n    makedirs(export_client_config['export_dir'], exist_ok=True)\n\n    if export_client_config['debug']:\n        print(url, token)\n    now = str(datetime.now())\n\n    if args.all:\n\t\t#kod migracji element\u00f3w \n\nif __name__ == '__main__':\n    main()<\/code><\/pre>\n\n\n\n
Kod wywo\u0142ania:<\/p>\n\n\n\n
python migrate_db.py --export_profile <profil \u017ar\u00f3d\u0142a> --import_profile <profil celu> --all --bypass-windows-check --azure<\/code><\/pre>\n\n\n\n
Dosz\u0142y wi\u0119c dwa nowe parametry<\/p>\n\n\n\n
export_profile, import_profile<\/p>\n\n\n\n
Parametry –bypass-windows-check i –azure s\u0105 specyficzne dla \u015brodowiska na kt\u00f3rym wykonywana jest migracja.<\/p>\n\n\n\n
Migracja u\u017cytkownik\u00f3w:<\/h3>\n\n\n\n
Aby z migrowa\u0107 u\u017cytkownik\u00f3w przygotowali\u015bmy poni\u017cszy kod w if args.all<\/strong>:<\/p>\n\n\n\n
if args.users:\n    print(\"Export all users and groups at {0}\".format(now))\n    e_scim_c = ScimClient(export_client_config)\n    start = timer()\n    # log all users\n    e_scim_c.log_all_users()\n    end = timer()\n    print(\"Complete Users Export Time: \" + str(timedelta(seconds=end - start)))\n    start = timer()\n    # log all groups\n    e_scim_c.log_all_groups()\n    end = timer()\n    print(\"Complete Group Export Time: \" + str(timedelta(seconds=end - start)))\n    \n    print(\"Import all users and groups at {0}\".format(now))\n    i_scim_c = ScimClient(import_client_config)\n    start = timer()\n    i_scim_c.import_all_users_and_groups()\n    end = timer()\n    print(\"Complete Users and Groups Import Time: \" + str(timedelta(seconds=end - start)))if args.users:\n            print(\"Export all users and groups at {0}\".format(now))\n            e_scim_c = ScimClient(export_client_config)\n            start = timer()\n            # log all users\n            e_scim_c.log_all_users()\n            end = timer()\n            print(\"Complete Users Export Time: \" + str(timedelta(seconds=end - start)))\n            start = timer()\n            # log all groups\n            e_scim_c.log_all_groups()\n            end = timer()\n            print(\"Complete Group Export Time: \" + str(timedelta(seconds=end - start)))\n            \n            print(\"Import all users and groups at {0}\".format(now))\n            i_scim_c = ScimClient(import_client_config)\n            start = timer()\n            i_scim_c.import_all_users_and_groups()\n            end = timer()\n            print(\"Complete Users and Groups Import Time: \" + str(timedelta(seconds=end - start)))<\/code><\/pre>\n\n\n\n
Brak zmian w pozosta\u0142ych klasach, wi\u0119c zwyk\u0142e po\u0142\u0105czenie export\/import wystarczy w tym przypadku.<\/p>\n\n\n\n
Migracja klastr\u00f3w:<\/h3>\n\n\n\n
Aby z migrowa\u0107 u\u017cytkownik\u00f3w przygotowali\u015bmy poni\u017cszy kod w if args.all<\/strong>:<\/p>\n\n\n\n
if args.clusters:\n    print(\"Export the cluster configs at {0}\".format(now))\n    e_cl_c = ClustersClient(export_client_config)\n    start = timer()\n    # log the cluster json\n    e_cl_c.log_cluster_configs()\n    e_cl_c.log_cluster_policies()\n    end = timer()\n    print(\"Complete Cluster Export Time: \" + str(timedelta(seconds=end - start)))\n    # log the instance pools\n    print(\"Start instance pool logging ...\")\n    start = timer()\n    e_cl_c.log_instance_pools()\n    end = timer()\n    print(\"Complete Instance Pools Export Time: \" + str(timedelta(seconds=end - start)))\n                \n    print(\"Import all cluster configs at {0}\".format(now))\n    i_cl_c = ClustersClient(import_client_config)\n    print(\"Start import of instance pool configurations ...\")\n    start = timer()\n    i_cl_c.import_instance_pools()\n    i_cl_c.log_instance_pools(log_file='instance_pools_new.log')\n    end = timer()\n    print(\"Complete Instance Pools Creation Time: \" + str(timedelta(seconds=end - start)))\n    print(\"Start import of cluster policies ...\")\n    start = timer()\n    i_cl_c.import_cluster_policies(log_pool_file='instance_pools_new.log')\n    end = timer()\n    print(\"Complete Cluster Policies Creation Time: \" + str(timedelta(seconds=end - start)))\n    print(\"Start import of cluster configurations ...\")\n    start = timer()\n    i_cl_c.import_cluster_configs()\n    end = timer()\n    print(\"Complete Cluster Import Time: \" + str(timedelta(seconds=end - start)))<\/code><\/pre>\n\n\n\n
W tym miejscu natrafiamy jednak na problem. Je\u015bli mamy klastry zdefiniowane u\u017cywaj\u0105c cluster pools lub cluster policies , natrafimy na problem migracji ID tych obiekt\u00f3w do nowego \u015brodowiska. Gdy\u017c, w nowym \u015brodowisku te elementy dostaj\u0105 nowe unikalne identyfikatory. Aby wi\u0119c poprawnie przenie\u015b\u0107 te elementy musimy zachowa\u0107 kolejno\u015b\u0107 tworzenia i przenoszenia obiekt\u00f3w, oraz zbudowa\u0107 map\u0119 migracji ID \u017ar\u00f3d\u0142owych do ID docelowych.<\/p>\n\n\n\n
Kroki:<\/p>\n\n\n\n
  • eksport konfiguracji cluster, cluster pool, cluster policies<\/li>
  • import zaczynamy od cluster pools, jednocze\u015bnie eksportuj\u0105c utworzone w tym kroku pool-e z nowego docelowego \u015brodowiska<\/li>
  • nast\u0119pnie import cluster policies dodaj\u0105c jako argument nowe cluster pool ustworzone krok wi\u0119cej<\/li>
  • nast\u0119pnie import cluster<\/li><\/ul>\n\n\n\n
    Modyfikacje import cluster policies:<\/h4>\n\n\n\n
    Poni\u017cej znajduje si\u0119 kod zmodyfikowany w klasie ClustersClient<\/strong>:<\/p>\n\n\n\n
    def import_cluster_policies(self, log_file='cluster_policies.log', acl_log_file='acl_cluster_policies.log', log_pool_file='instance_pools.log'):\n    policies_log = self.get_export_dir() + log_file\n    acl_policies_log = self.get_export_dir() + acl_log_file\n    policies_pool_log = self.get_export_dir() + log_pool_file\n    pool_map = dict()\n    if os.path.exists(policies_pool_log):\n        with open(policies_pool_log, 'r') as policy_pool_fp:\n            for pp in policy_pool_fp:\n                policy_pool_conf = json.loads(pp)\n                print(policy_pool_conf['default_tags']['DatabricksInstancePoolId']+':'+policy_pool_conf['custom_tags']['orig_instance_pool_id'])\n                pool_map[policy_pool_conf['custom_tags']['orig_instance_pool_id']] = policy_pool_conf['default_tags']['DatabricksInstancePoolId']\n    \n    # create the policies\n    if os.path.exists(policies_log):\n        with open(policies_log, 'r') as policy_fp:\n            for p in policy_fp:\n                policy_conf = json.loads(p)\n                # when creating the policy, we only need `name` and `definition` fields\n                for ld in pool_map.keys():\n                    policy_conf['definition'] = policy_conf['definition'].replace(ld,pool_map[ld])\n                \n                #print(policy_conf['definition'])\n                create_args = {'name': policy_conf['name'],\n                               'definition': policy_conf['definition']}\n                resp = self.post('\/policies\/clusters\/create', create_args)\n        # ACLs are created by using the `access_control_list` key\n        with open(acl_policies_log, 'r') as acl_fp:\n            id_map = self.get_policy_id_by_name_dict()\n            for x in acl_fp:\n                p_acl = json.loads(x)\n                acl_create_args = {'access_control_list': self.build_acl_args(p_acl['access_control_list'])}\n                policy_id = id_map[p_acl['name']]\n                api = f'\/permissions\/cluster-policies\/{policy_id}'\n                resp = self.put(api, acl_create_args)\n                print(resp)\n    else:\n        print('Skipping cluster policies as no log file exists')<\/code><\/pre>\n\n\n\n
    Polityki klastr\u00f3w zdefiniowane s\u0105 przy pomocy Json-a a jednym z argument\u00f3w jest ID cluster pool, je\u015bli wi\u0119c nie istnieje dany cluster pool w nowym \u015brodowisku , API zwr\u00f3ci b\u0142\u0105d. Musimy go wi\u0119c przemapowa\u0107. Znacznie lepszym rozwi\u0105zaniem by\u0142o by tutaj pos\u0142u\u017cenie si\u0119 nazw\u0105 cluster pool kt\u00f3re pomi\u0119dzy \u015brodowiskami nie ulegnie zmianie.<\/p>\n\n\n\n
    Podsumowanie:<\/h2>\n\n\n\n
    Za sob\u0105 mamy pierwsze modyfikacje kodu Databricks Labs, jednak nie jedyne. Dzi\u0119ki tym modyfikacjom mo\u017cemy zmigrowa\u0107 klastry i u\u017cytkownik\u00f3w wykonuj\u0105c poni\u017csz\u0105 komend\u0119.<\/p>\n\n\n\n
    python migrate_db.py --export_profile <profil \u017ar\u00f3d\u0142a> --import_profile <profil celu> --all --users --clusters --bypass-windows-check --azure<\/code><\/pre>\n\n\n\n
    Dalsze modyfikacje nied\u0142ugo.<\/p>\n","protected":false},"excerpt":{"rendered":"
    Wprowadzenie: Azure Databricks jest doskona\u0142\u0105 platform\u0105 do wykonania analiz , czy procesowania znacznej ilo\u015bci danych tzw. Big Data. Jednak nie jest to jeszcze do ko\u0144ca dojrza\u0142a platforma. Przez co cz\u0119sto pojawiaj\u0105 si\u0119 problemy zwi\u0105zane np. z procesem migracji danych\/klastr\u00f3w\/job\u00f3w\/kluczy\/notebook\u00f3w itp. do innych \u015brodowisk (takich jak Dev\/Test\/QA). Aby tego dokona\u0107 ka\u017cdy ze wspomnianych komponent\u00f3w migruje si\u0119 w inny spos\u00f3b. Jest jednak narz\u0119dzie przygotowane przez Databricks Labs, kt\u00f3re u\u0142atwia ten proces. Pami\u0119tajmy \u017ce Databricks jest platform\u0105 skupion\u0105 na podej\u015bciu API First, czyli \u017ce najpierw implementowany jest kod u\u017cywany w API, a nast\u0119pnie dobudowywane s\u0105 elementy takie jak GUI , klasy w python\/cli czy innych j\u0119zykach. Dzi\u0119ki temu mamy sami mo\u017cemy zbudowa\u0107\/rozbudowa\u0107 narz\u0119dzia migracyjne poszczeg\u00f3lnych komponent\u00f3w znaj\u0105c tylko definicj\u0119 API. Github: Kod migracji platformy Databricks Na stronie w readme znajduje si\u0119 do\u015b\u0107 rozbudowana dokumentacja jak u\u017cywa\u0107 kodu Python. My jednak dodamy co\u015b od siebie, tzn. opakujemy metody export\/import w jedn\u0105 klase do migracji konkretnych zasob\u00f3w. Implementacja w Azure Databricks: G\u0142\u00f3wne pliki wywo\u0142ywane w oryginalnej implementacji to export_db.py oraz import_db.py. Czyli operacje export\/import s\u0105 niezale\u017cne od siebie, co za tym idzie aby wykona\u0107 migracj\u0119 pomi\u0119dzy \u015brodowiskami, nale\u017cy wykona\u0107 seri\u0119 skrypt\u00f3w aby dane najpierw wyeksportowa\u0107 a nast\u0119pnie zaimportowa\u0107. My przygotowali\u015bmy migrate_db.py kt\u00f3ry wykonuje odpowiednie operacje […]<\/p>\n","protected":false},"author":1,"featured_media":15695,"comment_status":"closed","ping_status":"closed","sticky":false,"template":"","format":"standard","meta":[],"categories":[100,105,116],"tags":[99,127,107,128,125,126],"yoast_head":"\n\n\n\n\n\n\n\n\n\n\n\n\n\t\n\t\n\t\n\n\n\t\n\t\n\t\n