如何在自定义的PySpark ML流水线_transform()方法中创建一个Spark DataFrame？
在自定义的PySpark ML流水线_transform()方法中创建一个Spark DataFrame，可以按照以下步骤进行：
- 导入必要的PySpark模块：
from pyspark.sql import SparkSession
from pyspark.sql import DataFrame- 创建SparkSession对象：
spark = SparkSession.builder.getOrCreate()- 定义自定义的_transform()方法，并在方法中创建DataFrame：
def _transform(self, dataset: DataFrame) -> DataFrame:
# 创建自定义的DataFrame
custom_df = spark.createDataFrame([(1, 'A'), (2, 'B'), (3, 'C')], ['id', 'name'])
# 返回新的DataFrame
return custom_df在上述代码中，我们使用spark.createDataFrame()方法创建了一个自定义的DataFrame，该方法接受一个列表和一个列名列表作为参数，用于指定DataFrame的数据和列名。
- 将自定义的_transform()方法应用于流水线中的数据集：
# 假设pipeline是一个已定义的流水线对象
pipeline_model = pipeline.fit(input_data)
output_data = pipeline_model.transform(input_data)在上述代码中，我们使用pipeline.fit()方法拟合流水线模型，并使用pipeline_model.transform()方法将输入数据集转换为输出数据集。
这样，我们就在自定义的PySpark ML流水线_transform()方法中成功创建了一个Spark DataFrame。请注意，这只是一个示例，你可以根据实际需求进行修改和扩展。
相关·内容
”，第110行中，在包装器"/opt/module/spark-2.4.3-bin-hadoop2.7/python/lib/pyspark.zip/pyspark/ml/wrapper.py"，TypeError：init()中，有一个未预料到的关键字参数'ipnutCol‘异常，忽略在: Traceback (最近一次调用)：Fi
浏览 1提问于2019-08-24得票数 0
1回答
我试图在Spark2.4.0中创建一个自定义转换器。保存它很好。org.apache.spark.ml.util.令我困惑的是，等效的PySpark代码工作得很好：from pyspark.ml import Transformer, Default
浏览 0提问于2019-04-18得票数 2
回答已采纳
我有几个“无”值的数据框架。通过StringIndexer，将字符串列转换为浮动列后，"None“值被替换为number。谢谢。# Transform nominal attributes to numeric attributes
@staticmethod
def Trans
浏览 1提问于2018-04-29得票数 0
在Spark的中，转换器的transform()方法接受Spark DataFrame并返回DataFrame。我的自定义_transform()方法使用传入的DataFrame在处理之前创建一个RDD。这意味着在从_transform()返回之前，我的算法
浏览 4提问于2018-02-06得票数 0
回答已采纳
来自佛里安的示例代码|ball_column|keep_the |hall_column|| 6| 13| 20|代码的第一部分在禁用列表中删除列名代码的第二部分对列表中的特定列进行存储。在这个例子中，
浏览 0提问于2018-07-19得票数 12
回答已采纳
2回答
无法导入DSX环境中的spark
、、、、
我试过：import apache.spark.mllib.linalg.Vectors我也尝试过没有运气的pip inst
浏览 0提问于2018-04-01得票数 0
回答已采纳
我正在为我的皮斯喀特项目写我自己的变压器，我遇到了一个问题：from pyspark import keyword_only
from，所以我想把它们存储在一个单独的模块中(让我们称之为my_transformers.py，我想：“好吧，我需要一个SparkSession对象来完成这个工作……所以让我们把它放在__init__方法上吧.
浏览 1提问于2019-05-17得票数 0
回答已采纳
我有一个spark数据框，其中的" text“列有一些文本。我想要计算出现各种单词的行数-本质上是出现“术语”的“文档”的数量-以及相关的计数，如最频繁的单词，具有最频繁单词的行(或称为文档)。我正在使用来自pyspark.ml.feature的HashingTF。但是似乎找不到一种有效的方法来从输出中提取这些信息。f
浏览 25提问于2021-08-31得票数 0
回答已采纳
1回答
我有一个包含77列的数据集。我使用VectorAssembler将所有值组合在一起，然后使用MinMaxScaler对其进行缩放。]numVector是生成名为‘VectorAssembler’的列的缩放器，缩放器是MinMaxScaler%%time
from pyspark.ml\spark-3.0.1-bin-hadoop2.7\python\pyspark\ml
浏览 85提问于2021-03-01得票数 0
我正在尝试理解DataFrame列类型。当然，DataFrame不是一个物化的对象，它只是一组Spark的指令，将来要转换成代码。但我认为，这个类型列表代表了在执行操作时JVM中可能出现的对象类型。'pyspark.mllib.linalg.SparseVector'>
SparseVector <class 'pyspark.ml.linalg.Spars
浏览 1提问于2016-07-31得票数 7
回答已采纳
1回答
上下文：I有一个包含两列的数据框架: label和。org.apache.spark.sql.DataFrame = [label: int, features: vector]
其中的特性是使用mllib.linalg.VectorUDT构建的数值类型的问题：是否有向特征向量分配模式的方法？我想要跟踪每个功能的名称。userFeatures", attrs.asInstanceOf[Array[Attri
浏览 1提问于2016-02-10得票数 11
回答已采纳
1回答
我见过很多关于相似矩阵的堆叠溢出问题，但它们处理的是RDD或其他情况，我无法找到对我的问题的直接答案，于是我决定发布一个新的问题。from pyspark import SparkConf, SparkContext, SQLContext
from pyspark.ml.feature import VectorAssemblerfrom pyspark.ml.feature import Sta
浏览 1提问于2021-02-24得票数 6
回答已采纳
我正在使用星火MLLib进行预测，我想知道是否有可能创建您的自定义估计。以下是我希望我的模型与Spark所做的工作的可复制性import pandas as pdfrom <
浏览 0提问于2019-04-15得票数 6
在这里，您可以看到我到目前为止是如何做到的：model = svm.fit(train)
results = scores.select('probability',BinaryClassificationMetrics(scoreAndLabels)
pr
浏览 2提问于2018-01-25得票数 1
回答已采纳
5回答
如何使用spark-ml而不是spark-mllib处理分类数据？虽然文档不是很清楚，但似乎分类器(例如RandomForestClassifier、LogisticRegression )有一个featuresCol参数，它指定了DataFrame中的特性列的名称，还有一个labelCol参数，它指定了DataFrame中标记类的列的名称。显然，我希望在我的预测
浏览 10提问于2015-08-28得票数 49
1回答
火花放电中ML算法的训练
、、、
我对Pyspark还不熟悉，我试图在Pyspark中创建一个ML模型--我的目标是创建一个TFidf向量器并将这些特性传递给我的支持向量机模型。SparkContext(conf=conf) #dataset is a pandas dataframe()
tf = hashi
浏览 0提问于2018-02-26得票数 0
回答已采纳
1回答
如果我理解正确的话，spark中的ALS必须在训练前将user_id，item_id转换为整数？如果是这样，我现在能想到的唯一解决方案就是使用字典，并将每个user_id和item_id映射为整数，如
dictionary for item_id : {'18_34': 1, '18_35':2,但我想知道是否有其他优雅的方法来做到这一点？谢谢!
浏览 1提问于2015-11-16得票数 5
我正在尝试在Coursera“大数据的机器学习”的帮助下，在spark Mllib中实现一个决策树。我得到了下面的错误Traceback (most recent call last\python\lib\pyspark.zip\pyspark\ml\base.py&
浏览 12提问于2018-02-10得票数 0
回答已采纳
我有一个dataframe (df1)，其中包含m行和n列。我有另一个带有1行和n列的dataframe (df2)。如何用单行df1有效地计算每一行df2的点积？
浏览 2提问于2019-07-30得票数 0
回答已采纳
我正在使用pyspark(2.3.0) api来创建一个自定义转换器。我想要创建一个简单的转换器，它将任何功能作为param。我试着在TypeConverters中使用标识来实现它。密码有效。我面临的唯一问题是，我无法保存它，。它抛出函数对象不能被JSON序列化的错误。有办法绕开吗？
我在param中发送一个函数对象，因为我想用它来处理_transform方法中<
浏览 3提问于2020-05-12得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
云直播活动推荐
腾讯云开发者
