Questions tagged [pyspark]
The Spark Python API (PySpark) exposes the Apache Spark programming model to Python.
39,693
questions
0
votes
0
answers
11
views
iceberg is not a valid Spark SQL Data Source
I started a spark session in Python (I already installed pyspark and pyiceberg) like this
# Create a SparkSession
spark = SparkSession.builder.appName("MySQLRead") \
.config("spark....
0
votes
0
answers
4
views
Failed to find data source: kafka. Please deploy the application as per the deployment section of "Structured Streaming + Kafka Integration Guide
I was working on a kafka pyspark structured streaming workflow usind docker. All the dependencies are in docker: kafka,pyspark,postgres.
this pyspark script sshould receive the streams data from kafka ...
0
votes
1
answer
19
views
How to fill missing values in dataframe sequentially?
I have a data frame in PySpark that looks like this:
row_id, group_id
1, 1
2, null
3, null
4, null
5, 5
6, null
7, null
8, 8
9, null
10, null
11, ...
1
vote
1
answer
36
views
Filter out n number of duplicates
I have a dataframe such as follows:
id
first_name
last_name
sex
country
01
John
Doe
Male
USA
02
John
Doe
Male
Canada
03
John
Doe
Male
Mexico
04
Mark
Kay
Male
Italy
05
John
Doe
Male
Spain
06
...
0
votes
0
answers
7
views
How to Plot a Pareto or % Bar Chart in Pyspark Pandas?
I want to build a nice Pareto chart like this:
import pandas as pd
import matplotlib.pyplot as plt
from matplotlib.ticker import PercentFormatter
df = pd.DataFrame({'values': [1500.0, 300.0, 60.0, 40....
1
vote
0
answers
15
views
How to find documentation of dbruntime.dbutils.FileInfo class
I'm relatively new to Python, Spark and Cloud. I'm currently using the method dbutils.fs.ls(path), the databricks documentation doesn't specify that it returns a FileInfo object. When I do a "...
-9
votes
0
answers
48
views
how can I do exponentiation in SQL as in pyspark [closed]
I have the exponentiation 2.3174932900**252-1 written in Pyspark
which gives me the result below
9.652911279058692E91
How can I achieve the same in spark SQL?
0
votes
0
answers
8
views
Error while saving large Spark Dataframe to Hive table on Databricks
I'm trying to save spark dataframe as table in Databricks. The size of df is around 7GB.
df.write.format("delta").mode("overwrite").partitionBy("A","B","C&...
0
votes
0
answers
13
views
How to do conversion of pyspark RDD to dataframe?
RDD is generated after executing spark.SparkContext.parallelize(string.splitlines())
Later I tried toDF() then to pandas DF technique, partition, coalesce and various techniques all fail due to max ...
0
votes
0
answers
17
views
Empty DataSet after using join operation in Spark
I am trying to join multiple datasets one-by-one. The first join worked perfectly but the second one is giving me an empty DataFrame. As you can see from the output, the DataFrames are joined (in the ...
0
votes
1
answer
12
views
PySpark: How to Handle Inconsisten JSON Data Types for a Field
I have a large JSON string stored in a column in a delta table. I am using structured streaming for reads and writes. I have a pesky nested field 'value' which comes through in two types: string, ...
-1
votes
0
answers
14
views
Azure Datafactory API
I'm working on collecting information from my pipelines in the Data Factory API and creating a data map with this information. I'm using Databricks to access and gather this data, but the API has ...
0
votes
1
answer
16
views
Create&update a column if an other column in the same dataframe is the highest of its column
I am using the dataframe(dm1) as follows:
number pk
100 k1
200 k2
300 k3
400 k4
330 k3
500 k5
220 k2
370 k5
180 k2
90 k1
470 ...
1
vote
1
answer
18
views
spark onExecutorMetricsUpdate cannot get executor_id
In the code below, in pyspark 2.x, using executorMetricsUpdate.execId() always gave an executor id that was numeric, such as '31' (string type though). But after upgrading from pyspark 2.x to 3.4.1 ...
0
votes
0
answers
9
views
AWS Databricks: Table dropped when trying to overwrite with truncate flag set into Exasol
I am trying to write dataframe data into a table in Exasol from AWS Databricks using pyspark while truncating the data before the overwrite. I am using the follow command:
`_df.write.format("jdbc&...
1
vote
1
answer
50
views
How to run dictionary over file
I have got dataset like that
RespID
AGE
GEO
GEN
PROD
BRND
Q1
Q2
Q3
1
1
1
1
3
3
4
1
6
2
1
2
2
2
1
1
4
9
3
4
2
2
1
1
3
4
6
4
1
1
2
3
2
7
4
3
5
2
1
1
3
2
1
4
3
6
1
1
2
2
3
2
4
2
7
2
2
2
3
3
7
2
...
0
votes
0
answers
13
views
Shifted column when reading csv file in pyspark
I have a csv file stored in gcs, when reading this file some values are being shifted by 2 columns and I don't know why. One example is the following row:
202100000264277,01212022,*******2752,...
0
votes
0
answers
27
views
Handling JSON file in databricks using pyspark
I come across with a project that needs to load a lot of json files into azure data lake using databricks.
However, the data in these json files is not consistent.
For example
JSON File A:
{
"...
0
votes
1
answer
27
views
rlike not working for a snowpark dataframe
I am trying to fetch records having the string 'o' in a dataframe column but no records are showing up.
Currently in column 'Variant_nm' in dataframe I have
Junk_df1=df.filter(col('Variant_Nm')=='...
0
votes
1
answer
21
views
Pyspark dropDuplicates and drop_duplicates()
I have a data frame of only one column in pyspark, CUSTOMER_ID with repeated values. I want to drop duplicates.
It is a huge dataset, and I have used df.drop_duplicates() which worked. A colleague of ...
0
votes
1
answer
21
views
Pyspark DDL SQL Query on different host
I am new to python and entire gluescript thing. I wanted to execute a delete query on a different DB server. I am not sure what I am doing wrong
def remove_from_db(driver, url, dbtable, user, ...
0
votes
0
answers
28
views
Merging Cluster Lists from Multiple Algorithms: Identifying Methods and SQL Implementation
I'm facing a challenge in merging lists of clusters generated by different algorithms using the same set of data items, identifiable by their IDs. Each algorithm clusters these IDs based on varying ...
0
votes
0
answers
25
views
Spark UI - History Server - Unable to view driver logs
I have deployed "Spark History Server" in Azure Kubernetes Services(AKS) using Helm chart.
Then I submitted a SparkApplication(SparkOperatorOnk8s) with the below yaml
apiVersion: ...
0
votes
2
answers
41
views
How to perform average over months using window function with null values in between?
I have a dataframe like below
df = spark.createDataFrame(
[(1,1,10), (2,1,10), (3,1,None),(4,1,10),(5,1,10),(6,1,20) \
,(7,1,20), (1,2,10),(2,2,10),(3,2,10),(4,2,20),(5,2,20)],
["Month&...
0
votes
1
answer
25
views
Conditionally filling null values with left join in pyspark
I am trying to create a joined table using two tables, series and events. The structure for series table looks like following:
series_id
step
timestamp
x
y
a
1
2018-08-14T15:28:00-0400
2
3
...
...
....
0
votes
0
answers
10
views
housepower jdbc connection with pyspark
I am using housepower clickhouse jdbc driver as I want support for array type of data. I am using jdbc in pyspark. There are certain steps we need to follow which are listed in https://housepower....
0
votes
0
answers
7
views
How to solve a Weighting Problem with pyspark
I have matrix A of dimension [nXm] like
w1 w2 0
w3 0 w4
0 w5 0
a vector b with known value of size [mX1]:
10
5
3
and a vector c of size [nX1]
0
0
0
I want now to find on solution (finding ...
0
votes
1
answer
28
views
In pyspark, what is the difference between dlt.read_stream() and spark.readstream()?
I am tryting to understand In pyspark, what is the difference between dlt.read_stream() and spark.readstream()? can we replace dlt.read_stream() with spark.readstream() without affecting the data?
...
0
votes
0
answers
33
views
Slow changing dimension | SCD type 1 deleting rows from prior date data
I have implemented SCD type 1 using merge into statement in azure databricks.
When I am doing first load, it’s loading the data that is expected with some just say 5000 rows for ingest date 28 Nov, ...
1
vote
1
answer
45
views
Pyspark - grouping the description column details in an array
In the below sample dataset, I have two groups 'A' and 'B'. The 'Description' column contains the details associated with group 'A' and whenever an instance of group 'B' occurs, I need to add the ...
2
votes
0
answers
24
views
Incremental Inverse Document Frequency without storing the past information
I compute the tf-idf everyday in my pipeline using pyspark to evaluate the significance of a keyword in a specific document. This enables me to generate a summary for utilization in my machine ...
1
vote
1
answer
43
views
How to check the version of Spark Core, not pyspark
I am running a different version of pyspark and spark core (what's installed in the cluster), is there a way to print the version of spark core? everything so far gives me the version of pyspark. I ...
-1
votes
0
answers
22
views
How to use dateutils module in pyspark dataframe
def is_date(value):
try:
parser.parse(value)
return True
except ValueError:
return False
def sparkFindDataType(df):
is_date_udf = udf(lambda x : is_date(x), ...
0
votes
0
answers
27
views
Nested Explosion in PySpark
I'm trying to explode a very nested dataframe, which has nesting till 3-4 levels, and wanted to know how to explode in a optimized and precise manner!
Schema of the Nested DataFrame:
root
|-- uuid: ...
0
votes
0
answers
16
views
How to use dateutil parser in Pyspark dataframe
Actually, I need to iterate through each column of Pyspark dataframe and check for each rows. If I have date column with any type of date format, I need to identify it as date and its format. So I'm ...
1
vote
1
answer
31
views
Reading multiple videos in parallel with PySpark
I need to write a PySpark script in order to read in multiple video from mp4 files parallelly and then process it in PySpark. (These 4 video stream represent multiple RTSP video stream which I will ...
0
votes
0
answers
30
views
How to use Merge Into in Databricks (PySpark) given the source has null values to insert into a defined target schema (delta table)
Background: I am building a json parser that can take in any format of json and write it to a delta table with a schema that can update based on new data / new columns that come in.
I am trying to ...
0
votes
1
answer
60
views
ADF: Selecting from a json object that has attributes and values pivoted
One of the oracle REST APIs response is like below:
`{
"attributeId": 300000000227671,
"attributeName": "BUSINESS_UNIT&...
0
votes
0
answers
16
views
Subplots visualization in Databricks
How would one visualize several subplots from one table, using the Databricks visualization function? I've seen this in a presentation and can't figure out how to make it work.
This is how it looks ...
0
votes
0
answers
13
views
Pyspark stuck when converting rdd to dataframe
I got a freezing status when I tried to run very simple pyspark codes (Spark 3.5.0, Apple Silicon M1 Pro Chip). It used to work well with Spark 3.4.2. I thought it was version compatibility issue and ...
0
votes
0
answers
21
views
Issue with rolling 6 month filter in pyspark
I'm trying to write a SQL query in pyspark that will output the past rolling 6 months as one of the columns. I was able to successfully do this with the code below, but apparently there's a way to do ...
0
votes
0
answers
12
views
Step functions input to glue job
I have a state machine which will be triggered by event bridge as soon as an object is uploaded to s3 bucket. The stat machine will first call a lambda function to validate the input file and then ...
-1
votes
0
answers
27
views
error while importing koalas from databricks
import numpy as np
2 import pandas as pd
----> 3 import databricks.koalas as ks
File c:\Users\diego\anaconda3\Lib\site-packages\databricks\koalas\__init__.py:74
66 raise RuntimeError(
67 ...
0
votes
2
answers
20
views
Pyspark window that counts shows and no-shows of appointments with a day lag
I want to add two columns to a pyspark dataframe that calculate the sum of True or False count in the psdln.awz column and do this for each unique psyin_iden_rn number. However with the condition that ...
0
votes
0
answers
11
views
Hadoop Hive org.apache.hadoop.ipc.RpcException RPC response exceeds maximum data length
I am working in setting Remote PostgreSQL Database for Hive Metastore using
Hive 3.1.3
Hadoop 3.3.4
Spark 3.2.4
PostgresSQL 14.9
I could successfully use hive and hiveserver2 as it is shown below
...
0
votes
0
answers
37
views
Passing argument on SparkKubernetesOperator
I am using a spark with airflow, but not able to pass the arguments. I have tried multiple ways please suggest where right way to do this this.
dag.py file:
base_operator = SparkKubernetesOperator(
...
0
votes
1
answer
53
views
extracting a table value and list
I have a table, in one of the columns all the values are presented in lists. At the same time, in each row in this column there is a list in which there is only one value. how do I get this value out ...
0
votes
2
answers
28
views
Check if any of the values from list are in pyspark column's list
I have this problem with my pyspark dataframe, I created a column with collect_list() by doing normal groupBy agg and I want to write something that would return Boolean with information if at least 1 ...
0
votes
0
answers
28
views
Pyspark with Iceberg Catalog not found
I'm attempting to create a basic Iceberg table and query it using PySpark on my local Mac. However, I'm encountering an issue where my code is unable to locate the catalog that I previously created ...
0
votes
0
answers
42
views
How to convert pyspark df to python string?
I just want to show you the way "How to convert pyspark df to python string?".
Task: get a python string from a pyspark dataframe.
If you know how to make this easier, let me know!)
I got ...