Python-modulok importálása Git-mappákból vagy munkaterületfájlokból

A Python-kódot tárolhatja Databricks Git-mappákban vagy munkaterületi fájlokban, majd importálhatja a Python-kódot a DLT-folyamatokba. A Modulok Git-mappákban vagy munkaterületfájlokban való használatával kapcsolatos további információkért lásd: Python- és R-modulok használata.

Python-modul importálása DLT-folyamatba

Az alábbi példa az adathalmaz-lekérdezések Python-modulként történő importálását mutatja be munkaterületfájlokból. Bár ez a példa a folyamat forráskódjának tárolására szolgáló munkaterület-fájlok használatát ismerteti, használhatja egy Git-mappában tárolt forráskóddal is.

A példa futtatásához kövesse az alábbi lépéseket:

1. Kattintson az Azure Databricks-munkaterület oldalsávjában található és Munkaterületre, hogy megnyissa a munkaterület böngészőjét.

2. A munkaterület böngészője segítségével válasszon ki egy könyvtárat a Python-modulokhoz.

3. Kattintson a kijelölt könyvtár jobb szélső oszlopában, majd kattintson a > fájl létrehozásaparancsra.

4. Adja meg a fájl nevét, például clickstream_raw_module.py. Megnyílik a fájlszerkesztő. Ha egy modult szeretne létrehozni a forrásadatok táblázatba való beolvasásához, írja be a következőt a szerkesztőablakba:

   from dlt import *
   
   json_path = "/databricks-datasets/wikipedia-datasets/data-001/clickstream/raw-uncompressed-json/2015_2_clickstream.json"
   
   def create_clickstream_raw_table(spark):
     @table
     def clickstream_raw():
       return (
         spark.read.json(json_path)
       )
   

5. Ha olyan modult szeretne létrehozni, amely előkészített adatokat tartalmazó új táblát hoz létre, hozzon létre egy új fájlt ugyanabban a könyvtárban, adja meg a fájl nevét, például clickstream_prepared_module.py, és írja be a következőket az új szerkesztőablakba:

   from clickstream_raw_module import *
   from dlt import read
   from pyspark.sql.functions import *
   from pyspark.sql.types import *
   
   def create_clickstream_prepared_table(spark):
     create_clickstream_raw_table(spark)
     @table
     @expect("valid_current_page_title", "current_page_title IS NOT NULL")
     @expect_or_fail("valid_count", "click_count > 0")
     def clickstream_prepared():
       return (
         read("clickstream_raw")
           .withColumn("click_count", expr("CAST(n AS INT)"))
           .withColumnRenamed("curr_title", "current_page_title")
           .withColumnRenamed("prev_title", "previous_page_title")
           .select("current_page_title", "click_count", "previous_page_title")
       )
   

6. Ezután hozzon létre egy folyamatjegyzetfüzetet. Nyissa meg az Azure Databricks kezdőlapját, és válassza a Jegyzetfüzet létrehozásalehetőséget, vagy kattintson Új elemre az oldalsávon, és válassza Jegyzetfüzetlehetőséget. A jegyzetfüzetet a munkaterület böngészőben is létrehozhatja, ha a , majd a > jegyzetfüzet létrehozásaparancsra kattint.

7. Nevezze el a jegyzetfüzetét, és ellenőrizze, hogy Python az alapértelmezett nyelv.

8. Kattintson létrehozása gombra.

9. Írja be a példakódot a jegyzetfüzetbe.

   Jegyzet

   Ha a jegyzetfüzet modulokat vagy csomagokat importál egy munkaterület-fájl elérési útjáról vagy a jegyzetfüzet könyvtárától eltérő Git-mappák elérési útjáról, manuálisan kell hozzáfűznie az elérési utat a fájlokhoz a sys.path.append()használatával.

   Ha egy Git-mappából importál egy fájlt, az elérési út elé kell írni a /Workspace/-t. Például sys.path.append('/Workspace/...'). A /Workspace/ elérési útból való kihagyása hibát eredményez.

   Ha a modulok vagy csomagok ugyanabban a könyvtárban vannak tárolva, mint a jegyzetfüzet, akkor nem kell manuálisan hozzáfűznie az elérési utat. A Git-mappa gyökérkönyvtárából való importáláskor sem kell manuálisan hozzáfűznie az elérési utat, mert a gyökérkönyvtár automatikusan hozzá van fűzve az elérési úthoz.

   import sys, os
   # You can omit the sys.path.append() statement when the imports are from the same directory as the notebook.
   sys.path.append(os.path.abspath('<module-path>'))
   
   import dlt
   from clickstream_prepared_module import *
   from pyspark.sql.functions import *
   from pyspark.sql.types import *
   
   create_clickstream_prepared_table(spark)
   
   @dlt.table(
     comment="A table containing the top pages linking to the Apache Spark page."
   )
   def top_spark_referrers():
     return (
       spark.read.table("catalog_name.schema_name.clickstream_prepared")
         .filter(expr("current_page_title == 'Apache_Spark'"))
         .withColumnRenamed("previous_page_title", "referrer")
         .sort(desc("click_count"))
         .select("referrer", "click_count")
         .limit(10)
     )
   

   Cserélje le <module-path> az importálni kívánt Python-modulokat tartalmazó könyvtár elérési útvonalára.

10. Hozzon létre egy folyamatot az új jegyzetfüzet használatával.

11. A folyamat futtatásához kattintson a Folyamat részletei lapon a Startgombra.

A Python-kódot csomagként is importálhatja. A következő kódrészlet egy DLT-jegyzetfüzetből importálja a test_utils csomagot a dlt_packages könyvtárból a jegyzetfüzetével megegyező könyvtárban. A dlt_packages könyvtár tartalmazza a fájlokat test_utils.py és __init__.py, és test_utils.py határozza meg a függvényt create_test_table():

import dlt

@dlt.table
def my_table():
  return spark.read.table(...)

# ...

import dlt_packages.test_utils as test_utils
test_utils.create_test_table(spark)