Category "pyspark"

Pyspark list to rdd and split function inside map throws error

What is the issue with this code in pyspark raw_data = ["James,Smith,36636,M,3000", "Michael,Rose,40288,M,4000", "Robert,Williams,42114,M,4000", "M

How can i change Rdd to Vectors.dense pyspark

Im new to pyspark I need to change my rdd : tfidf.collect() output: [('fuel', 0.06190145817054232), ('months', 0.03095072908527116), ('lasting', 0.03095072908

Text is not interpreting the null in source dataframe(csv) as blank/null in the resultant dataframe(txt.)

I read the source data from a CSV file to a dataframe. Some columns have null values as seen in dataframe A. Now, dataframe B is the resultant data frame while

How to use join text with group by in Pyspark?

I have a pyspark dataframe id events a0 a-markets-l1 a0 a-markets-watch a0 a-markets-buy c7 a-markets-z2 c7 scroll_down a0 a-markets-sell b2 next_screen I am

Calculate MD5 hash CSV file in synapse notebook using pyspark

I want to calculate a hash (md5) from a .CSV in pyspark stationed in ADLS gen2. It seems that pyspark can't transform a df into bytes. import hashlib df = spar

Pyspark normal distribution

I'm trying to plot a normal distribution law with pyspark, does anyone have a syntax please? I tried with the numpy, matplotlib and scipy libraries but I still

spark: exec: "executor": executable file not found in $PATH: unknown

I am trying to do some calculus by using petastorm v0.11.4 in a docker container and minikube v1.25.2 As long as I run the process locally, everything works as

condition should be a Column dataframe PySpark

When using df_hdr_join.count() > 0 in when statement, it gives an error 'condition should be a Column'. I tried following. df_result = df.withColumn('NUM', w

How to form sparkstreaming startingOffsets dynamically in scala

I have a dataframe like below. This is a dynamic dataframe and will grow as more Topic fields are getting added. val ds = Seq(("T1",0,44), ("T1",1,54),

Pyspark - Find sub-string from a column of data-frame with another data-frame

I have two different dataframes in Pyspark of String type. First dataframe is of single work while second is a string of words i.e., sentences. I have to check

PySpark MLLib APproximate nearest neighbour search for multiple keys

I want to use ANN from PySpark. I have a DataFrame of 100K keys for which I want to perform top-10 ANN searches on an already transformed Spark DataFrame. But i

How to expand months in pyspark

I have data as below +-----+---------+----------+ | TYPE|DTIN_MNTH|DTOUT_MNTH| +-----+---------+----------+ | A| 2022-03| 2022-05| | B| 2022-04|

Calculate MAPE and apply to PySpark grouped Dataframe [@pandas_udf]

Goal: Calculate mean_absolute_percentage_error (MAPE) for each unique ID. y - real value yhat - predicted value Sample PySpark Dataframe: join_df +----------+--

How to use ODBC connection for pyspark.pandas

In my following python code I successfully can connect to MS Azure SQL Db using ODBC connection, and can load data into an Azure SQL table using pandas' datafra

Create column using Spark pandas_udf, with dynamic number of input columns

I have this df: df = spark.createDataFrame( [('row_a', 5.0, 0.0, 11.0), ('row_b', 3394.0, 0.0, 4543.0), ('row_c', 136111.0, 0.0, 219255.0), (

How do we optimise an incremental merge involving a very large target table (10 TB) and smaller incremental source table in a data lake environment?

I came across this question recently in one of the interviews and haven't been able to find a satisfying answer to this question. The incremental merge could co

Extract value from complex array of map type to string

I have a dataframe like below. No comp_value 1 [[ -> 10]] 2 [[ -> 35]] The schema type of column - value is. comp_value: array (nullable = tru

Perform sklearn DBSCAN on PySpark dataframe column

I have a Spark dataframe that looks like this: +-----+----------+--------+-----+ |key1 |date |variable|value| +-----+----------+--------+-----+ | A49|2022

How to assume a AWS role in pyspark

I am currently using spark 3.1, and I am using spark_context._jsc.hadoopConfiguration().set("fs.s3a.access.key", config.access_id) spark_context._jsc.hadoopConf

How do i create a single SparkSession in one file and reuse it other file

I have two py files com/demo/DemoMain.py com/demo/Sample.py In both of the above files i am recreating the SparkSession object , In Pyspark,how do i create a S