Questions tagged [pyspark]

The Spark Python API (PySpark) exposes the Apache Spark programming model to Python.

Filter by
Sorted by
Tagged with
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....
henryangminh's user avatar
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 ...
Louis_Curry's user avatar
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, ...
bensiu's user avatar
  • 24.9k
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 ...
Kruttika Swaminathan's user avatar
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....
noniguez's user avatar
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 "...
agusteando's user avatar
-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?
guigahbr's user avatar
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&...
ekdz__'s user avatar
  • 77
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 ...
myself's user avatar
  • 1
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 ...
Aman Tripathi's user avatar
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, ...
Cole1998's user avatar
-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 ...
Ricardo Lima's user avatar
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 ...
3Pac's user avatar
  • 29
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 ...
fersarr's user avatar
  • 3,459
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&...
Mohamed Gamal's user avatar
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 ...
lifeofthenoobie's user avatar
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,...
Jenny's user avatar
  • 385
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: { "...
asdcascz's user avatar
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')=='...
anurag verma's user avatar
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 ...
Himanshu Choubey's user avatar
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, ...
Ashish Choubey's user avatar
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 ...
Joey Baruch's user avatar
  • 4,400
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: ...
Tad's user avatar
  • 831
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&...
Ash's user avatar
  • 55
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 ... ... ....
Prabhjot Singh Rai's user avatar
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....
sdk's user avatar
  • 178
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 ...
Lazloo Xp's user avatar
  • 878
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? ...
Aditya Raj's user avatar
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, ...
Saurabh's user avatar
  • 107
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 ...
BiggieIt's user avatar
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 ...
aysebilgegunduz's user avatar
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 ...
fersarr's user avatar
  • 3,459
-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), ...
Shalini's user avatar
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: ...
Kranthi Kiran's user avatar
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 ...
Shalini's user avatar
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 ...
Lăng Khoa's user avatar
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 ...
dataengineeringhelp's user avatar
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&...
Advaitha's user avatar
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 ...
SWPerse's user avatar
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 ...
tonny2v's user avatar
  • 11
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 ...
PracticingPython's user avatar
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 ...
Srujana K's user avatar
-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 ...
Diego Alonzo's user avatar
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 ...
Mitchell's user avatar
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 ...
Ivan's user avatar
  • 119
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( ...
Shivam Gupta's user avatar
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 ...
alex4994's user avatar
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 ...
Nice Guy Eddie's user avatar
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 ...
shalama's user avatar
  • 1,163
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 ...
MsNobody242's user avatar

1
2 3 4 5
794