1回答
当更新数据库表时，我有一段要执行的python代码。我怎样才能为同样的事情写一个Mariadb触发器呢？
浏览 0提问于2016-10-20得票数 0
re Spark文档2.3：
registerJavaFunction(name，javaClassName，returnType=None)源
将Java用户定义函数注册为SQL函数。
除了名称和函数本身之外，还可以选择指定返回类型。当未指定返回类型时，我们将通过反射来推断它。
参数：
name -用户定义函数的名称
javaClassName - java类的全限定名
returnType -注册的Java函数的返回类型。该值可以是pyspark.sql.types.DataType对象，也可以是DDL格式的类型字符串。
我的问题是：
我想有一个大量的UDF库，为火花2.3+，都是
浏览 76提问于2018-08-11得票数 0
2回答
我正在尝试在python类中创建一个Spark-UDF。这意味着，类中的方法之一是UDF。我收到一个名为“PicklingError: can't serialize object: TypeError: can‘t pickle _MovedItems objects”的错误 环境: Azure Databricks。(DBR版本6.1 Beta)代码执行:在内置笔记本中。Python版本: 3.5 Spark版本: 2.4.4 我尝试在单独的单元格中定义类外部的UDF，UDF起作用了。我不想写这样的代码，我需要遵循OOP原则，并希望保持它的结构。我已经在谷歌上尝试了所有的东西，都
浏览 26提问于2019-10-16得票数 5
回答已采纳
我正在尝试获取调用函数的单元格的列号和行号，或者在Excel的函数代码中传递的单元格的列号和行号。 如果我在excel中调用一个用户定义的函数(UDF)，比如，=some_udf(A2:D2)和后端，也就是说，这个函数的定义是用python写的，我想要获取调用这个函数的单元格行号和列号，或者可能使用作为参数传递的范围来获取它们。我想不出该怎么做。 请帮帮我！或者，如果可以使用其他工具，如openpyxl等，请一定要告诉我。
浏览 260提问于2021-07-01得票数 0
2回答
我有两份文件。functions.py有一个函数，并从该函数创建一个pyspark udf。main.py会尝试导入该自定义项。但是，在访问functions.py中的函数时，main.py似乎遇到了问题。
functions.py：
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
def do_something(x):
return x + 'hello'
sample_udf = udf(lambda x: do_something(x), Strin
浏览 3提问于2017-10-04得票数 5
回答已采纳
我正在尝试使用PySpark创建随机生成的大型数据。下面的Scala代码基本上就是我想要的，但是我找不到一种在PySpark中复制seq.fill()的有效方法。
def randomStringGen(length: Int) = scala.util.Random.alphanumeric.take(length).mkString
val df = sparkContext.parallelize(Seq.fill(4000){(randomStringGen(4), randomStringGen(4), randomStringGen(6))}, 10).toDF("col
浏览 0提问于2021-08-06得票数 1
2回答
如何在PySpark SQL中实现用户定义的聚合函数？
pyspark version = 3.0.2
python version = 3.7.10
作为一个最小的例子，我想用一个UDAF替换AVG聚合函数：
sc = SparkContext()
sql = SQLContext(sc)
df = sql.createDataFrame(
pd.DataFrame({'id': [1, 1, 2, 2], 'value': [1, 2, 3, 4]}))
df.createTempView('df')
rv = sql.sql(
浏览 24提问于2021-03-09得票数 4
回答已采纳
我在PySpark中的向量列上使用UDF有困难，可以在这里说明如下：
from pyspark import SparkContext
from pyspark.sql import Row
from pyspark.sql.types import DoubleType
from pyspark.sql.functions import udf
from pyspark.mllib.linalg import Vectors
FeatureRow = Row('id', 'features')
data = sc.parallelize([(0, Vecto
浏览 2提问于2015-06-18得票数 2
回答已采纳
我正在尝试应用一个带有两个参数的pandas_udf。但是我得到了这个错误。首先，我尝试使用一个参数，它是可以的： from pyspark.sql.functions import pandas_udf, PandasUDFType
from pyspark.sql import SparkSession
from pyspark.sql.types import *
spark = SparkSession \
.builder \
.config('spark.cores.max', 100) \
.getOrCreat
浏览 24提问于2018-12-19得票数 3
我在Python语言中有一个有效的lambda函数，它可以计算dataset1中的每个字符串与dataset2中的字符串之间的最高相似度。在迭代期间，它将字符串、最佳匹配和相似度以及其他一些信息写入bigquery。没有返回值，因为该函数的目的是向bigquery数据集中插入一行。这个过程需要相当长的时间，这就是为什么我想使用Pyspark和Dataproc来加速这个过程。
将熊猫数据帧转换为spark很容易。我在注册udf时遇到了问题，因为它没有返回值，而pyspark需要一个返回值。此外，我不知道如何将python中的'apply‘函数映射到pyspark变体。因此，基本上我的问题
浏览 16提问于2019-07-19得票数 2
回答已采纳
1回答
我知道我可以注册一个UDFand函数，因为它可以在SQL查询中使用：
def example(s):
return len(s)
sqlContext.udf.register("example_udf", example)
spark.sql("SELECT example_udf(col) FROM data")
或者我可以用udf包装Python函数，这样就可以将它应用于dataframe：
from pyspark.sql.functions import udf
from pyspark.sql.types import IntegerType
浏览 1提问于2018-04-13得票数 0
回答已采纳
我已经编写了一个模块，其中包含了在PySpark DataFrames上工作的函数。它们对DataFrame中的列进行转换，然后返回一个新的DataFrame。下面是代码的一个示例，缩短为只包含其中一个函数：
from pyspark.sql import functions as F
from pyspark.sql import types as t
import pandas as pd
import numpy as np
metadta=pd.DataFrame(pd.read_csv("metadata.csv")) # this contains metad
浏览 6提问于2016-03-10得票数 7
回答已采纳
我是Spark初学者。我使用的是Python和Spark数据帧。我刚刚了解了用户定义函数(udf)，人们必须先注册才能使用它。问:在什么情况下，你想创建一个udf而不是一个简单的(Python)函数？
非常感谢!
浏览 0提问于2017-12-01得票数 2
1回答
过滤并不真的删除行？
、、、、
我的数据经过两个连续的过滤传递，每个都使用一个布尔值的UDF。第一个筛选移除列在某些广播字典中未作为键出现的所有行。第二个筛选对此字典与当前键相关联的值施加阈值。
如果我只在第一次筛选之后显示结果，那么包含“c”的行就不会像预期的那样出现在其中。但是，试图显示第二个筛选的结果会导致u'c‘的KeyError异常。
sc = SparkContext()
ss = SparkSession(sc)
mydict={ "a" : 4, "b" : 6 }
mydict_bc = sc.broadcast(mydict)
udf_indict=func.u
浏览 0提问于2018-01-27得票数 2
回答已采纳
我希望可以选择通过MySQL存储过程调用Python。
出于这个原因，我想知道我是否可以使用Python来编写一个可以从MySQL调用的UDF。
如果这不可能，我可以使用什么替代方法来实现这一点。
浏览 1提问于2011-04-25得票数 3
回答已采纳
1回答
在我写一个在hive中使用的udf之前，有一个基本问题。我想连接两个基于自定义UDF的表，UDF从表a和表b中获取一个参数。我见过UDF的例子，它从要连接的一个表中获取参数。从两个表中获取参数是否同样有效？
浏览 1提问于2013-08-04得票数 3
1回答
请查找以下代码：
import pandas as pd
from scipy.stats import norm
import pyspark.sql.functions as F
from pyspark.sql.functions import pandas_udf
import math
from pyspark.sql.functions import udf
from scipy.special import erfinv
# create sample data
df = spark.createDataFrame([
(1, 0.008),
(2, -1.2
浏览 1提问于2021-08-24得票数 0
1回答
我想知道，每次创建或更改某些记录时，调用用C编写的UDF (用户定义函数)时，会产生什么样的性能影响(假设假设，UDF代码本身不需要时间--我将自己对其进行优化)。
假设我有能够在200 k写/s上运行SSD持久化命名空间的硬件，那么每次运行UDF时，我能期望至少有50k写/s吗？
子问题:什么会限制UDF的性能(上下文切换？)
问这个问题的原因是，国际航空公司正在使用这些UDF，例如用于大型数据类型，但根据工作人员的说法，它们的性能并不高(与KVS-操作相比)。我的用法是使用UDF在Redis集群中保持范围广泛的辅助索引的最新更新，允许更丰富的实时查询(例如，5-10个二级索引的交叉/联合)
浏览 1提问于2015-05-24得票数 1
回答已采纳
考虑到monetdbe是一个Python包，我乐观地认为Python定义的函数是可能的，但是我还没有找到一个例子。我试过这个：
drop function every_other_letter;
create function every_other_letter(s string)
returns string
language python {
letters = "".join(s[::2])
return letters
};
select every_other_letter('foobarbaz');
我知道这个错误：
Par
浏览 10提问于2022-01-18得票数 2
回答已采纳
我有一个pyspark数据帧
a = [
(0.31, .3, .4, .6, 0.4),
(.01, .2, .92, .4, .47),
(.3, .1, .05, .2, .82),
(.4, .4, .3, .6, .15),
]
b = ["column1", "column2", "column3", "column4", "column5"]
df = spark.createDataFrame(a, b)
现在，我想基于以下条件创建一个新列
df.withColu
浏览 0提问于2020-09-25得票数 1
1回答
我有一个包含列(id、id2、vec、vec2)的表，其中id是整数，向量是pyspark SparseVeectors。我想写一个udf来接受vec和vec2的点积，如下所示： def dot_product(vec, vec2): #or do the row and I can access them later as row.vec, row.vec2?
return vec.dot(vec2)
udf = udf(dot_product, FloatType())
dot_product = df.withColumn('dot_product', udf(
浏览 16提问于2020-07-08得票数 1
1回答
无法使用大型字典的Pyspark
、、、、
我有一个字典，由键= word，value = Array组成，由300个浮点数组成。我无法在我的“火花放电”中使用这本词典。当这本字典的大小是200万键时，它就不起作用了。但是当我把尺寸缩小到200 K时，它就能工作了。
这是我要转换为UDF的函数的代码。
def get_sentence_vector(sentence, dictionary_containing_word_vectors):
cleanedSentence = list(clean_text(sentence))
words_vector_list = np.zeros(300)# 300 dim
浏览 0提问于2019-08-19得票数 1
回答已采纳
根据这一汇合点页面：
https://cwiki.apache.org/confluence/display/FLINK/FLIP-106%3A+Support+Python+UDF+in+SQL+Function+DDL
python位于Flink 1.11中，可用于SQL函数中。
我在这里看了flink的医生：
https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/sqlClient.html
并在终端上尝试此操作，并使用以下参数启动SQLclient.sh：
$ sql-client.sh embedded --
浏览 1提问于2020-07-09得票数 0
我有一个Python脚本，我正在尝试将其转换为pyspark，我有一个在Python中运行在pd.DataFrame上的函数func，并返回一个列表列表。
例如：
[[14],[2,3]]
在python中，我最终在一个分组的pd.Dataframe上运行它，并且是这样运行的：
data.groupby('CUSTOMER_ID').apply(func)
输出是一个索引为CUSTOMER_ID的pd.Series，每个元素都是与每个CUSTOMER_ID相关的列表的列表
例如：
CUSTOMER_ID
123 [[14], [2, 3]]
浏览 30提问于2020-03-03得票数 0
回答已采纳
我目前在将Python Pandas函数转换为Python PySpark时遇到了一个问题，因为它们是不同的库。我想要做的是有一个查询函数，然后将它应用回相同的列。
这是我为Python Pandas所做的(Age是我试图从中检索的数据集中的列)：
Age = [1, 3, -100, -99999, 39, 60, 87, 20, 21, 77777]
def clean_age(Age):
if Age>=0 and Age<=95:
return Age
else:
return np.nan
df['Age&
浏览 17提问于2020-05-18得票数 0
回答已采纳
2回答
我编写了以下代码，运行良好。但是我想将UDF连接起来，这样就可以在几行代码中进行压缩。请建议我怎么做。下面是我写的代码。
val myUdf1 = udf((Number: Long) => ((Number) >> 24) & 255)
val myUdf2 = udf((Number: Long) => ((Number) >> 16) & 255)
val myUdf3 = udf((Number: Long) => ((Number) >> 8) & 255)
val myUdf4 = udf((Number
浏览 0提问于2020-06-17得票数 1
回答已采纳
有几个人(、、)讨论过在PySpark应用程序中使用Scala UDF，通常是出于性能原因。相反，我感兴趣的是在Scala Spark项目中使用python UDF。
我特别感兴趣的是使用sklearn (和)构建一个模型，然后有效地将其应用于Spark streaming作业中的记录。我知道我也可以在中的REST API和后面托管python模型，但是管理该任务的并发性和为托管模型设置API并不是我非常感兴趣的事情。
如果没有像Py4J这样的东西进行太多的定制开发，这是可能的吗？这只是一个坏主意吗？
谢谢!
浏览 0提问于2018-08-19得票数 11
1回答
我在服务器上安装了jython jar和Pig。还有猪缸。有人能帮我找出正确的步骤来捆绑它们，以便我可以使用我的Python UDF吗？谢谢
浏览 0提问于2013-03-07得票数 2
我在无线传感器网络上工作，为此，我创建了一个MySQL数据库，我在其中保存了传感器的状态，每当传感器的状态发生变化时，它就会在该数据库中插入一行新的状态和时间。然后，我使用python创建了一个接口，我希望在该接口中显示传感器的状态，并且每当MySQL数据库中的状态发生更改时，它都会在接口中(实时地)更改。所以我想知道，当该值发生变化时，是否可以从MySQL数据库向我的python应用程序发送通知?？
浏览 1提问于2019-04-19得票数 2
我想了解udf在火花放电中的工作原理。每次我们在dataframe之上使用udf时，python都会打开吗？
浏览 0提问于2018-11-26得票数 1
回答已采纳
1回答
我想知道我们是否可以在运行时将Scala中定义的方法转换为Python中的方法。我正在学习Apache Spark Scala API，作为我的项目的一部分，并寻找一种方法来将Scala方法转换为Python语言可识别的格式。我的目标是将方法从Scala程序发送到外部Python程序，并且Python程序应该能够使用或执行它。
Scala代码=> Lambda
line => line.split(" ")
有没有办法把它转换成Python可以理解的代码？
上面的例子绝对是非常基础和幼稚的。但是可以有任何可能的方法(考虑到Python也有函数支持的泛型方法)，是否有
浏览 0提问于2017-10-19得票数 1
2回答
我有如下所示的数据。我想按device分组，在每个组内按start_time排序。然后，对于组中的每一行，从其前面3行的窗口(包括其自身)中获得最频繁发生的站点。
columns = ['device', 'start_time', 'station']
data = [("Python", 1, "station_1"), ("Python", 2, "station_2"), ("Python", 3, "station_1"), ("Py
浏览 1提问于2022-01-13得票数 1
回答已采纳
1回答
我有一个名为‘count’的列的dataframe，我想将一个自定义函数"do_something“应用于列的每个元素，即每个数组。我不想修改dataframe，我只想做一个单独的操作与列计数。列的所有数组都具有相同的大小。
+----------------------+---------------------------------------+
|id| counts|
+----------------------+---------------------------------------+
|1| [8.0, 2.0, 3.0
浏览 2提问于2017-09-22得票数 1
回答已采纳
2回答
毫升起火花放电作用
、、、、
我对火种和蟒蛇有点陌生。我正在尝试将ML函数作为pyspark运行。
下面是一个示例：
from pyspark.sql.functions import col, pandas_udf
from pyspark.sql.types import StringType
df = spark.createDataFrame(['Bob has a dog. He loves him'], StringType())
def parse(text):
import spacy
import neuralcoref
nlp = spacy.load(
浏览 2提问于2020-06-16得票数 5
我有一个用scala写的dataframe a2： val a3 = a2.select(printme.apply(col(“PlayerReference”))) 列PlayerReference包含一个字符串。 调用udf函数的方法： val printme = udf({
st: String =>
val x = new JustPrint(st)
x.printMe();
}) 此udf函数调用java类： public class JustPrint {
private String ss =
浏览 1提问于2020-12-03得票数 0
回答已采纳
1回答
我有一个数据帧，我想要在其中使用pysaprk中的polyline库获取给定地理位置的lat_long +-----------------+--------------------+----------+
| vid| geolocations| trip_date|
+-----------------+--------------------+----------+
|58AC21B17LU006754|eurnE||yqU???????...|2020-02-22|
|2T3EWRFV0
浏览 30提问于2020-08-31得票数 0
回答已采纳
1回答
我试图写这样一个简单的星星之火。当我在星火版10.4.x-scala2.12上的databricks笔记本上测试它时。同样的代码运行得很好。当我在一个打包的jar中运行这个程序并以相同的星火版本向databricks提交时，它会导致这样的异常，
例外:在spark.sql(sql_stat).show(false)
Job aborted due to stage failure.
Caused by: NoClassDefFoundError: Could not initialize class com.test.TestClass$
:
:
at com.test.TestClass$.
浏览 10提问于2022-12-03得票数 0
1回答
我是pyspark的新手，我想把我现有的pandas / python代码转换成PySpark。
我希望对我的dataframe进行子集，以便只返回'original_problem'字段中包含特定关键字的行。
下面是我在PySpark中尝试的Python代码：
def pilot_discrep(input_file):
df = input_file
searchfor = ['cat', 'dog', 'frog', 'fleece']
df = df[df['orig
浏览 0提问于2018-05-18得票数 3
回答已采纳
1回答
我想让hive Python UDF除了来自列的数据之外，还接受常量参数。
如何才能做到这一点？
在下面的示例中，我希望下面的配置单元查询能够接受参数'10000'，并且my_udf应该能够解析该参数'10000'
# hive>
ADD FILE my_udf.py;
SELECT TRANSFORM(col1, col2, '10000') USING 'python my_udf.py' AS (udf_output) FROM my_table;
和
# my_udf.py
import sys
for line_s
浏览 1提问于2020-05-28得票数 0
1回答
假设我有一个SparseVector，我想对它的值求和，例如
v = SparseVector(15557, [3, 40, 45, 103, 14356], np.ones(5))
v.values.sum()
5.0
这样做效果很好。现在我想通过udf来做同样的事情，因为我有一个包含SparseVector列的DataFrame。在这里我得到了一个我不理解的错误：
from pyspark.sql import functions as f
def sum_vector(vector):
return vector.values.sum()
sum_vector_udf =
浏览 0提问于2017-11-03得票数 1
import pygeohash as pgh
pgh.encode(45,55)
'tpzpgxczbzur'
以上步骤非常有效。下面我试图创建一个数据框架：
l = [(45,25),(75,22),(85,20),(89,26)]
rdd = sc.parallelize(l)
geoCords = rdd.map(lambda x: Row(lat=x[0], long=int(x[1])))
geoCordsSchema = sqlContext.createDataFrame(geoCords)
geoCordsSchema.show()
+---+----
浏览 2提问于2017-12-02得票数 3
回答已采纳
2回答
我需要创建一个UDF，用于python，它使用java对象进行内部计算。
如果它是一条简单的python，我会做这样的事情：
def f(x):
return 7
fudf = pyspark.sql.functions.udf(f,pyspark.sql.types.IntegerType())
并使用以下方法进行调用：
df = sqlContext.range(0,5)
df2 = df.withColumn("a",fudf(df.id)).show()
但是，我所需要的函数的实现是用java而不是python实现的。我需要以某种方式包装它，这样我就可以用类似的
浏览 2提问于2016-03-23得票数 11
回答已采纳
我的输入是 'Apple-Pie-1024' 所需输出为 'Apple-Pie' 我可以使用python rsplit()方法做到这一点，如下所示 s.rsplit('-',1)[0].strip() 这将为我提供所需的输出 如何使用spark sql获得类似的输出？提前感谢
浏览 27提问于2021-07-26得票数 0
回答已采纳
3回答
如果这个问题用词不当，我很抱歉:我正在进行一个大规模的机器学习项目，我不喜欢用Java编程。我喜欢用Python编写程序。我听说过关于小猪的好话。我想知道是否有人可以向我澄清，在数学相关的工作中，Pig与Python的组合是如何使用的。另外，如果我要写“流式python代码”，Jython会出现在图片中吗？如果它真的出现在图片中，它是不是更有效率？
谢谢
附注:出于几个原因，我不喜欢使用Mahout的代码。我可能想要使用他们的一些数据结构:知道这是否可能做到这一点将是有用的。
浏览 3提问于2011-07-08得票数 5
我定义了一个pandas udf函数，并希望将除pandas.Series或pandas.DataFrame之外的其他参数传递给udf函数。我想使用partial函数来做这件事，但它出错了。我的代码如下： from functools import partial
from pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSession
from pyspark.sql.functions import pandas_udf, PandasUDFType
conf = SparkConf().setM
浏览 15提问于2019-12-18得票数 0
回答已采纳
1回答
如果我用sqlite3.create_function在python中创建了一个UDF，我可以从SQLite命令行调用这个函数吗？
浏览 3提问于2016-03-18得票数 0
回答已采纳
2回答
是否可以在pyspark应用程序的UDF中使用广播数据帧。
我的代码调用了下面所示的电火花数据Code中的广播数据。
fact_ent_df_data =
sparkSession.sparkContext.broadcast(fact_ent_df.collect())
def generate_lookup_code(col1,col2,col3):
fact_ent_df_count=fact_ent_df_data.
select(fact_ent_df_br.TheDate.between(col1,col2),
浏览 0提问于2018-10-29得票数 0
回答已采纳
1回答
我正在尝试一些与pyspark相关的实验，在连接到AWS EMR实例的jupyter笔记本上。我有一个spark dataframe，它从s3读取数据，然后过滤掉一些东西。使用df1.printSchema()输出打印模式，如下所示： root
|-- idvalue: string (nullable = true)
|-- locationaccuracyhorizontal: float (nullable = true)
|-- hour: integer (nullable = true)
|-- day: integer (nullable = true)
|-- da
浏览 39提问于2019-05-09得票数 2
回答已采纳
1回答
我在上有一个大约6亿行的表。我有一个python进程，它通过sqlalchemy_redshift连接发出以下查询：
begin;
UPDATE dogs
SET computed_dog_type = f_get_dog_type(name, breed, age, color)
WHERE week = :week;
commit;
该查询运行良好；但是，一次运行超过6亿行的速度太慢。where条款有效地限制了每周的行数，其值从2K到230万。
我将查询代码包装在python线程中，执行如下所示：
16:38 $ python dog_classifier.py update_range
浏览 0提问于2016-11-02得票数 2
回答已采纳
我想在BigQuery中的特定数据集中创建一个UDF。我尝试过使用create_table和来自Python 的其他API方法，但似乎没有一种方法有效。
是否有任何方法或方法将UDF (存储在文件中并以字符串形式读取)上载到BigQuery中？
浏览 2提问于2020-10-01得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
腾讯会议活动推荐
腾讯云开发者
