Questions tagged [apache-spark-sql]
Apache Spark SQL is a tool for "SQL and structured data processing" on Spark, a fast and general-purpose cluster computing system. It can be used to retrieve data from Hive, Parquet etc. and run SQL queries over existing RDDs and Datasets.
26,796
questions
0
votes
1
answer
13
views
ParseException: mismatched input '2022' expecting {<EOF>, ';'}
I have the following PySpark/SQL code:
extracted_at = "2022-01-31 00:00:00"
extracted_at_date = "2022-01-31 00:00:00"
current_fiscal_period = (sqlContext.sql('SELECT CUR_PER FROM ...
0
votes
3
answers
26
views
How to use ILIKE in PySpark?
I have a query
select * from
table
where (primary_product NOT IN ('No Technical Enforcement')
or group_name ilike ('%stove%'))
I want to convert the same query into PySpark SQL, but I am not ...
0
votes
2
answers
49
views
Convert Nested Array Using PySpark
I want to perform some aggregations on this table using PySpark. My table name is COMPLEX_DATA and is stored in Snowflake. The COORDINATES column's datatype is VARIANT. The nested arrays are not of ...
0
votes
1
answer
10
views
Why did Spark interchange values of two columns?
Please can someone explain why spark interchanges the values of two columns when querying a DataFrame?
The values of ProposedAction are returned for SimpleMatchRate vise versa.
Here is the code sample:...
0
votes
0
answers
27
views
Pyspark dataframe saving incorrectly to parquet?
I'm running into a perplexing problem when saving a dataframe to a parquet file after performing a series of transformations.
After I run a series of transformation steps, I check that the final ...
-1
votes
0
answers
18
views
Combining numerical and text features in (deep) neural networks using pyspark ml package
How to combine numerical and text features in (deep) neural networks as stated here using pyspark? Any examples/references would be really helpful. As far as I have searched online, I haven't been ...
0
votes
3
answers
39
views
update set values with dictionary with multiple values per key
I have a sqldb table looking like this below
| tag | unit | description|
|:---- |:------:| -----:|
| 1a | % | mass flow |
| 2a | * | head flow |
and I want to update values into the table with ...
0
votes
0
answers
18
views
catching EsHadoopIllegalArgumentException in pyspark
I have to catch the exception EsHadoopIllegalArgumentException (org.elasticsearch.hadoop.EsHadoopIllegalArgumentException) in a pyspark script. Don't understand how to import that exception in order ...
0
votes
0
answers
24
views
Data in delta table is not changing with overwriting option?
I am creating a delta table on top of a dataframe with external location. I was trying to overwrite delta table but data in delta table is getting overwrite but underlined data in delta lake gen2 is ...
2
votes
1
answer
56
views
OOM using Spark window function with 30 days interval
I have this data frame:
df = (
spark
.createDataFrame([
[20210101, 'A', 103, "abc"],
[20210101, 'A', 102, "def"],
[20210101, 'A', 101, "def"],
[...
-1
votes
1
answer
48
views
+50
Spark is inconsistent with unusually encoded CSV file
Context:
As part of data pipeline, I am working on some flat CSV files.
Those files have unusual encoding and escaping rules.
My intention is too preprocess those and convert to parquets for ...
0
votes
1
answer
34
views
Changing Multiple Values in PySpark Dataframe
The sample of the dataset I am working on:
test = sqlContext.createDataFrame([(1,2),
(1,3),
(4,5)],
...
0
votes
0
answers
30
views
Writing tables in Azure Data Lake Gen 2 takes too long with Pyspark Notebook
I am quite new in programming with parallel processing with Spark in Azure Synapse by using PySpark notebook. I need to do some transformations of data stored in Cosmos DB and then save the results in ...
1
vote
1
answer
43
views
How to repartition spark dataframe into smaller partitions
I have a dataframe which is partitioned by date.
In normal processing, I am processing a week of data at a time, so this means I have 7 partitions. I would like to increase this number of partitions, ...
0
votes
0
answers
17
views
AnalysisException: Path does not exist: While running github code using colab
I am running a github code for signature detection in google colab and in the code image is read as binary file for further processing using spark. The image is inside a folder which is in the same ...
1
vote
1
answer
42
views
Spark/Scala - Join resultset giving type mismatch error while performing withColumn operation; found : org.apache.spark.sql.Column required: Boolean
I have two dataframes that I am trying to join and based on the joined set trying to assign a flag column.
demandDF1
+--------+-----------+---------+
|rgn_nm |file_crt_dt|file_vrsn|
+--------+--------...
-2
votes
1
answer
40
views
Pyspark conditional running window
I have a dataframe which looks like this
starttime | endtime |
2022-01-01 03:25:53 | 2022-01-01 03:25:52 |
2022-01-01 03:25:53 | 2022-01-01 03:25:52 |
2022-01-01 03:25:53 | 2022-...
1
vote
1
answer
20
views
Not able to perform operations on resulting dataframe after "join" operation in PySpark
df=spark.read.csv('data.csv',header=True,inferSchema=True)
rule_df=spark.read.csv('job_rules.csv',header=True)
query_df=spark.read.csv('rules.csv',header=True)
join_df=rule_df.join(query_df,rule_df....
1
vote
2
answers
45
views
Adding None to PySpark array
I want to create an array which is conditionally populated based off of existing column and sometimes I want it to contain None. Here's some example code:
from pyspark.sql import Row
from pyspark.sql ...
1
vote
1
answer
31
views
How to aggregate on multiple columns using SQL or spark SQL
I have following table:
Id col1 col2
1 a 1
1 b 2
1 c 3
2 a 1
2 e 3
2 f 4
Expected output is:
Id col3
1 a1b2c3
2 a1e3f4
The aggregation computation involves ...
0
votes
0
answers
17
views
Why is a Spark dataframe with 1 large row taking a long time to write?
I have a dataframe with just 1 row, but that row is about 1 gigabyte in size. Writing it takes a very long time. My job runs for about an hour because of that, is that normal?
I did coalesce(10) ...
1
vote
2
answers
30
views
Losing data formats when saving Spark dataframe to delta table in Azure Databricks
When I try to save a Spark Dataframe as delta table to the Azure Databricks Delta Lake, all my data types turn into strings. Additionally all the dates in what used to be a datetime type column get ...
0
votes
1
answer
19
views
Grouping alternative items with PySpark
The sample of the dataset I am working on:
# Creating the DataFrame
test =sqlContext.createDataFrame([(1,2),(2,1),
(1,3),(2,3),
(3,2),(3,1)...
0
votes
0
answers
31
views
No TypeTag available for String
I'm trying to run my fat jar using scala -classpath "target/scala-2.13/Capstone-assembly-0.1.0-SNAPSHOT.jar" src/main/scala/project/Main.scala, but I get an error caused by .toString: val ...
1
vote
5
answers
71
views
Split column with JSON string to columns each containing one key-value pair from the string
I have a data frame that looks like this (one column named "value" with a JSON string in it). I send it to an Event Hub using Kafka API and then I want to read that data from the Event Hub ...
-1
votes
1
answer
29
views
Pyspark read all files and write it back it to same file after transformation
Hi I have files in a directory
Folder/1.csv
Folder/2.csv
Folder/3.csv
I want to read all these files in a pyspark dataframe/rdd and change some column value and write it back to same file.
I have ...
0
votes
0
answers
20
views
Timestamp format changes when writing file in csv format in spark
I am writing my data to my local using spark write
below is the code
spark_raw_5.repartition(1).write.mode("overwrite").option("header","true")
.option("escape",...
0
votes
1
answer
11
views
For spark-phoenix, I don't want to pull the whole data from HBase. Is there any way to pull the data from table according to my condition?
It is too slow to pull the whole data from Hbase. I only need to process data for one hour.
So I want to know whether I can specify the SQL in config and pull one-hour data like Spark's support to ...
0
votes
1
answer
50
views
Extract json data inside array to rows in Spark
I'm working with Azure Databricks to parse over a large number of small (4kb) JSON files containing information from a remote monitoring station with the following schema:
root
|-- Body: struct (...
-1
votes
1
answer
45
views
Counting files in a directory and its sub-directories created after a specified timestamp in PySpark
In my PySpark code, I need to count all avro files created in any sub-directories of a given directory after a specified timestamp and store that count in a variable.
Any recommendations/examples how ...
-1
votes
0
answers
16
views
Pyspark: FileNotFoundError: [WinError 2] The system cannot find the file specified
Python code generating different error:
spark = SparkSession.builder.appName('PySpark DataFrame From RDD').getOrCreate()
FileNotFoundError: [WinError 2] The system cannot find the file specified
0
votes
1
answer
22
views
Scala spark data frame phone number validation
I want to perform some validation on column Phone number mentioned below and then update the "Correct" column with "Y" (when the number looks valid) or "N" (when invalid)....
0
votes
1
answer
19
views
Find the time difference after subtracting weekend time
ID
START_TIME_
END_TIME_
TIME_DIFF
TIME_SPENT_WORKING_DAYS
1
2022-04-18 09:58:38.906
2022-04-18 12:50:27.204
10308298
2
2022-04-18 21:26:25.671
2022-04-18 21:26:36.313
10642
3
2022-04-18 21:27:12....
1
vote
2
answers
34
views
How can I close db connection from spark.read?
I can select data from database in Spark like this:
var df = spark.read.
format("jdbc").
option("url", "jdbc:db://<DB server>:<DB port>/<dbname>").
...
0
votes
1
answer
29
views
PYSPARK - join nullsafe on multiple columns
Let's assume we are having two dataframes, which we want to compare for differences with a leftanti join:
data1 = [
(1, 11, 20, None),
(2, 12, 22, 31),
]
data2 = [
(1, 11, 20, None),
(2, 12, ...
0
votes
0
answers
22
views
How to type cast an Array as empty String array
I can type cast NULL as a string.
How can I type cast an empty array as an empty array of strings?
I need to solve it inside the SQL query.
The following snippet throws a ValueError: Some of types ...
0
votes
1
answer
33
views
How to create a Spark dataframe with timestamp?
How can I create this Spark dataframe with timestamp data type in one step using python? Here is how I do it in two steps. Using spark 3.1.2
from pyspark.sql.functions import *
from pyspark.sql.types ...
0
votes
1
answer
23
views
Resolved attribute(s) ... missing from
I will start with a code dump (self-contained minimal but representative example), followed by the output and afterwards will provide the problem statement.
With that out of the way: code dump (...
0
votes
0
answers
17
views
what's the difference/advantange between DSL(dataframe API) ans SQL in spark sql [closed]
I'm new to Bigdata/Spark sql/Hive, recently I am learning to use spark sql.
From my understanding by now, to operate dataframe, we can have below two way, which will lead the same result data dfFinal,...
3
votes
1
answer
42
views
Groupby column and create lists for other columns, preserving order
I have a PySpark dataframe which looks like this:
Id timestamp col1 col2
abc 789 0 1
def 456 ...
0
votes
3
answers
42
views
Add a column from a function of 2 other columns in PySpark
I have two columns in a data frame df in PySpark:
| features | center |
+----------+----------+
| [0,1,0] | [1.5,2,1]|
| [5,7,6] | [10,7,7] |
I want to create a function which calculates the ...
0
votes
1
answer
35
views
Migration from Spark 2.4.0 to Spark 3.1.1 caused SortMergeJoin to change to BroadcastHashJoin
I'm currently working on a Spark migration project that aims to migrate all Spark SQL pipelines for Spark 3.x version and take advantage of all performance improvements on it. My company is using ...
0
votes
1
answer
22
views
How to extract and filter on nested struct fields in PySpark?
Here's the data schema
df = spark.read.parquet(link)
df.printSchema()
# root
# |--user_id : long
# |--date : string
# |--totals: struct
# | |--time: long
# | |--views: long
# |...
0
votes
1
answer
30
views
How to chain explode and struct field selection?
The dataframe:
from pyspark.sql import functions as F
df = spark.createDataFrame([([(1, 2), (3, 4)],)], 'col_name array<struct<c1:int,c2:int>>')
df.show()
# +----------------+
# | ...
0
votes
1
answer
39
views
Count occurrences of a filtered value in a window
I have the following dataframe:
| Timestamp | info |
+-------------------+----------+
|2016-01-01 17:54:30| 8 |
|2016-02-01 12:16:18| 2 |
|2016-03-01 12:17:57| 1 |
|...
0
votes
1
answer
29
views
Pyspark Dataframe Iterate Array Columns
In PySpark, I have a dataframe I'm trying to parse multiple columns with arrays. The last two rows in the dataframe contains multiple values I would like to parse into separate rows.
+-------------+--...
1
vote
1
answer
27
views
Parse JSON column then add new column in the same dataframe with parsed value
Here is an example of what I'm trying to accomplish:
Case:
column1
column2
json_column
One
Two
{'A': '1', 'B': '2', 'C': '3'}
Desired output:
column1
column2
json_column
B
One
Two
{'A': '1', 'B': ...
0
votes
1
answer
32
views
aggregate pyspark dataframe and create multiple columns
I have a dataframe that looks like below
+-----------+--------------+----------------+--------------------+-----------------------+
|CUSTOMER_ID|mkt_channel_id|mkt_channel_name|mkt_channel_category|...
1
vote
1
answer
27
views
Apache spark : How deep is the comparison of rows in RDD or DF
I want to understand the behavior of DF.intersect().
so the question came to mind, especially when we have complex Rows having complex fields. (deep tree)
-2
votes
0
answers
25
views
attributeerror: 'GBTRegressionModel' object has no attribute 'summary' [closed]
I want to find metrics like RMSE, R2 score etc but without summary I can't do it on training set, instead I am able to predict test data set and then using regression evaluator RegressionEvaluator of ...