Spark Python作業開發流程

1. 準備Spark Python作業

2. 打包Spark Python作業

3. 上傳Spark Python作業

4. 提交Spark Python作業

步驟一：準備Spark Python作業

1. 下載Spark Python作業示例壓縮包，下載鏈接為Spark Python作業示例。

2. 解壓Spark Python作業示例壓縮包，解壓后的目錄名稱為lindorm-spark-examples。打開lindorm-spark-examples/python目錄，參考python目錄結構。

3. 項目開發的根目錄以your_project為例，介紹項目的目錄結構。

   1. 在your_project目錄下新建__init__.py文件，內容為空。

   2. 改造入口文件。

      1. 在入口文件中編寫代碼，將your_project添加到sys.path中，代碼詳情請參見Spark Python作業示例中的lindorm-spark-examples/python/your_project/main.py文件的Notice1部分。

         # Notice1: You need to do the following step to complete the code modification:
         # Step1: Please add a "__init__.py" to your project directory, your project will act as a module of launcher.py
         # Step2: Please add current dir to sys.path, you should add the following code to your main file
         current_dir = os.path.abspath(os.path.dirname(__file__))
         sys.path.append(current_dir)
         print("current dir in your_project: %s" % current_dir)
         print("sys.path: %s \n" % str(sys.path))

      2. 在入口文件中將入口邏輯封裝到main(argv)方法中。代碼詳情請參見Spark Python作業示例中的lindorm-spark-examples/python/your_project/main.py文件的Notice2部分。

         # Notice2: Move the code in `if __name__ == "__main__":` branch to a new defined main(argv) function,
         # so that launcher.py in parent directory just call main(sys.argv)
         def main(argv):
             print("Receive arguments: %s \n" % str(argv))
         
             print("current dir in main: %s \n" % os.path.abspath(os.path.dirname(__file__)))
             # Write your code here
         
         
         if __name__ == "__main__":
             main(sys.argv)

   3. 創建Spark Python作業啟動的入口文件，用來調用main(argv)方法。在根目錄your_project創建同級目錄launcher.py，可以復制Spark Python作業示例中的lindorm-spark-examples/python/launcher.py文件。

步驟二：打包Spark Python作業

1. 打包項目依賴的Python環境和第三方類庫。推薦使用Conda或者Virtualenv將依賴類庫打包為tar包，具體操作請參見Python Package Management。

   重要

   • 使用Conda或者Virtualenv打的tar包通過spark.archives傳遞，可以是spark.archives支持的所有格式。詳細說明，請參見spark.archives。

   • 請在Linux環境下完成該操作，以保證Lindorm計算引擎能正常識Python二進制文件。

2. 打包項目文件。將your_project文件打包為.zip或者.egg格式文件。

   • 執行以下命令將項目文件打包為.zip格式文件。

     zip -r project.zip your_project

   • 將項目文件打包為.egg格式文件，具體操作請參見Building Eggs。

步驟三：上傳Spark Python作業

步驟四：提交Spark Python作業

Python作業開發示例

1. 下載并解壓Spark Python作業示例。

2. 改造入口文件，打開Python目錄下的your_project/main.py文件并配置相關代碼。

   1. 將your_project目錄添加到sys.path中。

      current_dir = os.path.abspath(os.path.dirname(__file__))
      sys.path.append(current_dir)
      print("current dir in your_project: %s" % current_dir)
      print("sys.path: %s \n" % str(sys.path))

   2. 在main.py文件中添加入口邏輯，如下示例初始化SparkSession。

      from pyspark.sql import SparkSession
      spark = SparkSession \
          .builder \
          .appName("PythonImportTest") \
          .getOrCreate()
      print(spark.conf)
      spark.stop()

3. 在Python目錄下打包your_project文件。

   zip -r your_project.zip your_project

4. 在Linux環境下，使用Conda打包Python運行環境。

   conda create -y -n pyspark_conda_env -c conda-forge numpy conda-pack
   conda activate pyspark_conda_env
   conda pack -f -o pyspark_conda_env.tar.gz

5. 將打包好的your_project.zip和pyspark_conda_env.tar.gz上傳至OSS，并將Python目錄下的launcher.py文件上傳至OSS。

6. 通過以下兩種方式提交作業。

   • 通過Lindorm控制臺提交作業，操作方法請參見通過控制臺管理作業。

   • 通過DMS提交作業，操作方法請參見通過DMS管理作業。

作業診斷

Python作業提交成功后，可以在作業列表頁面查看作業運行狀況和SparkUI訪問地址，具體操作請參見查看作業。作業提交過程中如果有其他問題，請提交工單并將JobID和WebUI地址提供給工單處理人員。