大数据开发治理平台 DataWorks：Notebook 进阶开发

开发环境调试

1. 挂载数据集：进入个人开发环境详情页，在存储配置 > 数据集中配置数据集。

2. 在代码中访问：数据集将被挂载到个人开发环境的挂载路径下，在代码中直接读写该路径。

   # 假设已在个人开发环境设置了挂载路径为/mnt/data/dataset的数据集
   import pandas as pd
   
   # 直接使用挂载路径
   file_path = '/mnt/data/dataset/testfile.csv'
   df = pd.read_csv(file_path)
   
   # 使用 PyODPS 将数据写入 MaxCompute
   o = %odps
   o.write_table('mc_test_table', df, overwrite=True)
   print(f"成功将数据写入 MaxCompute 表 mc_test_table。")

生产环境部署

1. 挂载数据集：在 Notebook 节点编辑页面的右侧导航栏，进入调度配置 > 调度策略，添加上述数据集。

2. 在代码中访问：提交发布之后，数据集将被挂载到生产环境的挂载路径下，在代码中直接读写该路径。

   # 假设已在个人开发环境设置了挂载路径为/mnt/data/dataset的数据集
   import pandas as pd
   
   # 直接使用挂载路径
   file_path = '/mnt/data/dataset/testfile.csv'
   df = pd.read_csv(file_path)
   
   # 使用 PyODPS 将数据写入 MaxCompute
   o = %odps
   o.write_table('mc_test_table', df, overwrite=True)
   print(f"成功将数据写入 MaxCompute 表 mc_test_table。")

引擎对比

MaxCompute Spark

通过 Livy 连接到 MaxCompute 项目内置的 Spark 引擎。

1. 建立连接：在 Python 单元格中运行以下命令。系统将自动创建或复用 Spark 会话。

   # 创建Spark Session，并停止Livy。
   %maxcompute_spark

2. 执行 PySpark 代码：连接成功后，在新的 Python 单元格中，使用 %%spark Cell Magic 执行 PySpark 代码。

   # 在使用MaxCompute Spark时，Python单元格中必须以%%spark作为首行代码
   %%spark
   
   df = spark.sql("SELECT * FROM your_mc_table LIMIT 10")
   df.show()

3. 手动释放连接：在开发调试结束后，可手动停止或删除会话。在生产环境运行时，系统将自动停止并删除当前任务实例的Livy，无需手动处理。

   # 清理Spark Session，并停止Livy。
   %maxcompute_spark stop
   
   # 清理Spark Session，并停止Livy，最后删除Livy
   %maxcompute_spark delete

EMR Serverless Spark

通过连接您预先创建的 Livy Gateway 与 EMR Serverless Spark 交互。

1. 建立连接：在执行命令前，需在单元格右下角选择要连接的 EMR 计算资源 和 Livy Gateway。

   # 基础连接
   %emr_serverless_spark
   
   # 或在连接时传入自定义 Spark 参数，注意需要自定义Spark参数时，需要写两个百分号
   %%emr_serverless_spark
   {
     "spark_conf": {
       "spark.emr.serverless.environmentId": "<EMR Serverless Spark 运行环境ID>",
       "spark.emr.serverless.network.service.name": "<EMR Serverless Spark 网络连接ID>",
       "spark.driver.cores": "1",
       "spark.driver.memory": "8g",
       "spark.executor.cores": "1",
       "spark.executor.memory": "2g",
       "spark.driver.maxResultSize": "32g"
     }
   }

   说明

   自定义参数与全局配置的关系

   • 默认行为：您在此处定义的自定义参数仅对本次连接（Session）生效，是一次性的。如果您未提供自定义参数，系统将自动使用在管理中心中配置的全局参数。

   • 推荐用法：对于需要在多个任务或被多人复用的配置，建议在管理中心 > Serverless Spark > SPARK参数中进行全局配置，以保证一致性并方便统一管理。

   • 优先级规则：当同一个参数在自定义参数和全局配置中都被设置时，哪个最终生效，取决于管理中心中的全局配置是否优先选项：

     • 勾选：全局配置将覆盖本次的自定义参数。

     • 不勾选：本次的自定义参数将覆盖全局配置。

2. （可选）重新连接：若Livy gateway页面的token被管理员误删后可通过该命令重新创建。

   # 重新连接，刷新当前个人开发环境的livy token
   %emr_serverless_spark refresh_token

3. 执行 PySpark 或 SQL 代码：连接成功后，内核已切换。您可以在 Python 单元格 中直接编写 PySpark 代码，或在 EMR Spark SQL 单元格 中编写 SQL。

   1. 通过EMR Spark SQL向计算资源提交并执行SQL代码

      经过%emr_serverless_spark成功建立连接后，可在EMR Spark SQL Cell中，直接写SQL语句，无需在单元格中选择计算资源。

      EMR Spark SQL Cell将复用%emr_serverless_spark的连接，提交至目标计算资源中运行。

      

   2. 通过Python向计算资源提交并执行PySpark代码

      通过 %emr_serverless_spark 成功建立连接后，可以在新的Python单元格中提交和执行PySpark代码，无需在单元格中添加%%spark前缀。

      

4. 手动释放连接

   重要

   在多人共享一个 Livy Gateway 的情况下，`stop` 或 `delete` 命令会影响所有正在使用该网关的用户，请谨慎操作。

   # 清理Spark Session，并停止Livy。
   %emr_serverless_spark stop
   
   # 清理Spark Session，并停止Livy，最后删除Livy
   %emr_serverless_spark delete

AnalyticDB for Spark

通过创建 Spark Connect Server 连接到 AnalyticDB for Spark 引擎。

1. 建立连接：为确保网络连通，必须在连接参数中正确配置交换机 ID 和安全组 ID。 在执行命令前，需在单元格右下角选择 ADB Spark 计算资源。

   # 必须配置交换机ID和安全组ID以建立网络连接
   %adb_spark add \
    --spark-conf spark.adb.version=3.5 \
    --spark-conf spark.adb.eni.enabled=true \
    --spark-conf spark.adb.eni.vswitchId=<ADB的交换机ID> \
    --spark-conf spark.adb.eni.securityGroupId=<个人开发环境的安全组ID>

   如何查找交换机ID和安全组ID？

   • 交换机ID (vswitchId)：前往阿里云AnalyticDB MySQL控制台，在实例详情页查看网络信息的交换机ID。

   • 安全组ID (securityGroupId)：进入个人开发环境详情页的网络设置中所选安全组，查看以sg-开头的安全组ID。

     重要

     为确保网络连通，创建个人开发环境时，建议选择与您的AnalyticDB for Spark实例相同的VPC和交换机。

2. 执行 PySpark 代码：连接成功后，在新的 Python 单元格中，执行 PySpark 代码。

   # 只能对 C-Store 表执行操作
   df = spark.sql("SELECT * FROM my_adb_cstore_table LIMIT 10")
   df.show()

   说明：AnalyticDB for Spark 引擎当前只能处理具有 'storagePolicy'='COLD' 属性的 C-Store 表。

3. 手动释放连接：在开发环境调试结束后，手动清理连接会话以节约资源。生产环境运行时，系统将自动清理资源。

   %adb_spark cleanup