标签: databricks

我在Databricks中使用Spark 2.1版.我有一个数据框wamp,我想要添加一个名为的列region,该列应该采用常量值NE.但是,NameError: name 'lit' is not defined当我运行以下命令时出现错误:

wamp = wamp.withColumn('region', lit('NE'))

我究竟做错了什么？

我正在尝试将数据从spark数据帧导出到.csv文件:

df.coalesce(1)\
  .write\
  .format("com.databricks.spark.csv")\
  .option("header", "true")\
  .save(output_path)

它正在创建一个文件名"part-r-00001-512872f2-9b51-46c5-b0ee-31d626063571.csv"

我希望文件名为"part-r-00000.csv"或"part-00000.csv"

在AWS S3上创建文件时,我对如何使用os.system命令的限制.

如何在保留文件中的标题的同时设置文件名？

谢谢!

我正在使用Databricks Connect从IntelliJ IDEA（Scala）在本地Azure Databricks群集中运行代码。

一切正常。我可以在IDE中本地连接，调试和检查。

我创建了一个Databricks作业来运行我的自定义应用程序JAR，但由于以下异常而失败：

19/08/17 19:20:26 ERROR Uncaught throwable from user code: java.lang.NoClassDefFoundError: com/databricks/service/DBUtils$
at Main$.<init>(Main.scala:30)
at Main$.<clinit>(Main.scala)

我的Main.scala类的第30行是

val dbutils: DBUtils.type = com.databricks.service.DBUtils

就像本文档页面上的描述一样

该页面显示了一种访问在本地和群集中均可使用的DBUtil的方法。但是该示例仅显示了Python，而我正在使用Scala。

以既可以使用databricks-connect在本地工作又可以在运行JAR的Databricks作业中工作的方式访问它的正确方法是什么？

更新

似乎有两种使用DBUtils的方法。

1）这里描述的DbUtils类。引用文档，此库允许您构建和编译项目，但不能运行它。这不允许您在集群上运行本地代码。

2）此处描述了 Databricks Connect 。这使您可以在Databricks集群中运行本地Spark代码。

问题在于这两种方法具有不同的设置和程序包名称。似乎没有一种在本地使用Databricks Connect的方法（在群集中不可用），但是随后通过sbt / maven添加了使用DbUtils类的jar应用程序，以便群集可以访问它。

我创建了一个 DataFrame，我想将其写入/导出到表中的 Azure DataLake Gen2 旁边（需要为此创建新表）。

将来我还需要使用新的 DataFrame 更新此 Azure DL Gen2 表。

在 Azure Databricks 中，我创建了一个连接 Azure Databricks -> Azure DataLake 来查看我的文件：

感谢如何在 Spark / pyspark 中编写它的帮助。

谢谢你！

我正在运行运行时 8.1（包括 Apache Spark 3.1.1、Scala 2.12），试图让 hyperopt 按定义工作

https://docs.databricks.com/applications/machine-learning/automl-hyperparam-tuning/hyperopt-spark-mlflow-integration.html ​

py4j.Py4JException: Method maxNumConcurrentTasks([]) does not exist

当我尝试

spark_trials = SparkTrials()

我需要做什么特别的事情才能使其正常工作吗？

这是我正在使用的集群

{
    "autoscale": {
        "min_workers": 1,
        "max_workers": 2
    },
    "cluster_name": "mlops_tiny_ml",
    "spark_version": "8.2.x-cpu-ml-scala2.12",
    "spark_conf": {},
    "aws_attributes": {
        "first_on_demand": 1,
        "availability": "SPOT_WITH_FALLBACK",
        "zone_id": "us-west-2b",
        "instance_profile_arn": "arn:aws:iam::112437402463:instance-profile/databricks_instance_role_s3",
        "spot_bid_price_percent": 100,
        "ebs_volume_type": "GENERAL_PURPOSE_SSD",
        "ebs_volume_count": 3,
        "ebs_volume_size": 100
    },
    "node_type_id": "m4.large",
    "driver_node_type_id": "m4.large",
    "ssh_public_keys": [],
    "custom_tags": {},
    "spark_env_vars": {},
    "autotermination_minutes": 120,
    "enable_elastic_disk": false,
    "cluster_source": "UI",
    "init_scripts": [],
    "cluster_id": "0xxxxxt404"
}

这是我正在使用的代码 https://docs.databricks.com/applications/machine-learning/automl-hyperparam-tuning/hyperopt-model-selection.html

我试图找出为什么在使用 Databricks Job API 时出现以下错误。

{ "error_code": "INVALID_PARAMETER_VALUE", "message": "集群验证错误：缺少必填字段：settings.cluster_spec.new_cluster.size" }

我做了什么：

1. 我使用 Databricks UI 创建了一个在单节点集群上运行的作业。
2. 我从 UI 复制并粘贴了作业配置 json。
3. 我删除了我的作业，并尝试通过使用作业 API 发送 POST 以及复制的 json 来重新创建它，如下所示：

{
    "new_cluster": {
        "spark_version": "7.5.x-scala2.12",
        "spark_conf": {
            "spark.master": "local[*]",
            "spark.databricks.cluster.profile": "singleNode"
        },
        "azure_attributes": {
            "availability": "ON_DEMAND_AZURE",
            "first_on_demand": 1,
            "spot_bid_max_price": -1
        },
        "node_type_id": "Standard_DS3_v2",
        "driver_node_type_id": "Standard_DS3_v2",
        "custom_tags": {
            "ResourceClass": "SingleNode"
        },
        "enable_elastic_disk": true
    },
    "libraries": [
        {
            "pypi": {
                "package": "koalas==1.5.0"
            }
        }
    ],
    "notebook_task": {
        "notebook_path": "/pathtoNotebook/TheNotebook",
        "base_parameters": {
            "param1": "test"
           
        }
    }, …

对 Databricks 中的索引如何工作感到好奇。您能否将分区视为索引，因为它有效地将数据组织在分组的子类别中？

我试图将 parquet 文件转换为 Excel 文件。但是，当我尝试使用 pandas 或openpyxl引擎这样做时，它显示“ Operation not supported”错误。但是，我可以使用databricks 中的openpyxl引擎读取 excel 文件。

在阅读以下代码时，它正在工作：

xlfile = '/dbfs/mnt/raw/BOMFILE.xlsx'
tmp_csv = '/dbfs/mnt/trusted/BOMFILE.csv'
pdf = pd.DataFrame(pd.read_excel(xlfile, engine='openpyxl'))
pdf.to_csv (tmp_csv, index = None, header=True)

但是，当我尝试使用 openpyxl 和 xlswriter 编写相同的内容时，它不起作用：

parq = '/mnt/raw/PRODUCT.parquet'
final = '/dbfs/mnt/trusted/PRODUCT.xlsx'
df = spark.read.format("parquet").option("header", "true").load(parq)
pandas_df = df.toPandas()
pandas_df.to_excel(final, engine='openpyxl')
#pandas_df.to_excel(outfile, engine='xlsxwriter')#, sheet_name=tbl)

我得到的错误：

FileCreateError: [Errno 95] Operation not supported

OSError: [Errno 95] Operation not supported
During handling of the above exception, another exception occurred: …

我已经设置了 Amundsen，并且 UI 工作正常。我正在尝试运行其存储库中的示例中给出的示例 Delta Lake 加载程序。

"""
This is a example script for extracting Delta Lake Metadata Results
"""

from pyhocon import ConfigFactory
from pyspark.sql import SparkSession

from databuilder.extractor.delta_lake_metadata_extractor import DeltaLakeMetadataExtractor
from databuilder.job.job import DefaultJob
from databuilder.loader.file_system_neo4j_csv_loader import FsNeo4jCSVLoader
from databuilder.models.table_metadata import DESCRIPTION_NODE_LABEL
from databuilder.publisher import neo4j_csv_publisher
from databuilder.publisher.neo4j_csv_publisher import Neo4jCsvPublisher
from databuilder.task.task import DefaultTask

# NEO4J cluster endpoints
NEO4J_ENDPOINT = 'bolt://localhost:7687/'

neo4j_endpoint = NEO4J_ENDPOINT

neo4j_user = 'neo4j'
neo4j_password = 'test'
cluster_key = 'my_delta_environment'
database = 'delta'
# Or …

我有一个笔记本，它将处理该文件并以结构化格式创建数据框。现在我需要导入在另一个笔记本中创建的数据框，但问题是在运行笔记本之前我需要验证仅适用于我需要运行的某些场景。

通常要导入所有数据结构，我们使用%run。但就我而言，它应该是 if 子句和 then notebook run 的组合

if "dataset" in path": %run ntbk_path

它给出错误“路径不存在”

if "dataset" in path": dbutils.notebook.run(ntbk_path)

这个我无法获得所有的数据结构。

有人可以帮我解决这个错误吗？