You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
I guess that is just for separation. It would mean that we want the spark-sql-kafka-0-10_2.12 package for the 3.1.2 version of Spark. So let's say you are running Spark 3.1.1. Then the last value after : will change to 3.1.1.
Thanks a lot! I actually have another unrelated question. I have been struggling to find out how EventSim and Kafka work together to create the 4 topics: listen_events, page_view_events, auth_events, status_change_events. Is that hard-coded and only works for the Million Song Dataset? I also saw that each event has different columns, so you assign different Spark schemas for them.
Right. Eventsim is currently coded to generate events for a music streaming website. You can change that but that is written in Scala. So if you have experience with that, why not. The topics are also currently hardcoded. You can check the codebase here
From the confluent control center UI. It should be available on port 9021 of your VM. You will have to forward that port to your local machine and open it in the browser.
No worries. If you check the youtube video I put out a few days back, you should get an idea of how to navigate through the project. You'll find the link in the README
Hi. Do you mind answering another question of mine? I hope this is not bothering you.
When I set the environmental variable KAFKA_ADDRESS to be the external IP, the eventsim somehow couldn't create all 4 topics. The logs suggested that it struggles to find the correct IP address. But when I don't set the environmental variable (and thus setting it to localhost by default as per your code), then it works. Do you have any idea why?
I am not sure why this would happen. Eventsim is running on the same VM as Kafka, hence it should pick up directly via that. Eventsim doesn't refer to the KAFKA_ADDRESS variable as such. Have you mentioned the correct IP, and do you have your 9092 port open?
My mistake. It was opened automatically. (Sorry I deleted the comment after I realized that). Now it works. But let me try again after setting the env variable.
Yes, it does. If you want to connect from an external VM, like the Spark cluster in our case, you need to write to the external IP address instead of the localhost. Hence I added the KAFKA_ADDRESS variable in the docker-compose.yml.
resource "google_compute_firewall" "port_rules" {
project = var.project
name = "kafka-broker-port"
network = var.network
description = "Opens port 9092 in the Kafka VM for Spark cluster to connect"
allow {
protocol = "tcp"
ports = ["9092"]
}
source_ranges = ["0.0.0.0/0"]
target_tags = ["kafka"]
}
I didn't follow everything in your project so I missed this part. Is this the reason why it doesn't work? But without this step, it does open up port 9092 on my VSCode when I SSH to the Kafka VM though. Or they are totally different things?
So on VS Code, you are forwarding the port. It is not the same as opening the port. You are able to forward the port to your local computer because you already authenticated when you did an SSH. On the other hand, you open your port to allow connections from some other VM in the network. By default, no ports are accessible to any other VMs inside or outside the network. You need to specify, which port would you like to open to accept connections.
I have another question. I see that you set up a Dataprocs server for Spark streaming. Do you think it makes sense if I do everything on a single VM and hence serve it locally? So I was thinking everything would involve "localhost" and the master node would be "local[*]" etc.
You can. However, In the real-world Spark would always be running on some kind of a cluster. So to make things run on the cluster would be a wiser choice if you plan to showcase the infra choices in an interview or somewhere else.
By the way, I am not sure this is a mistake on my behalf, but I think I might have found some potential bugs:
The KAFKA_ADDRESS for the Kafka docker-compose YAML file is unnecessary since both Kafka and the EventSim container are in the same virtual machine. Kindly refer to this link.
The relevant change I made is the removal of the environmental variable
Otherwise, I couldn't make it work. (Ignore my reply a few days ago, I too thought it worked when I set the env variable to the external IP of the VM, but it actually did not work. My bad.)
Hey, apologies for the late reply. There's a lot going on at this point and I kinda missed this comment.
I had to start writing the message to external IP instead of localhost explicitly so that my Spark Streaming process can connect and stream the messages. Earlier I had that same line as yours in the YAML file. I had to then add the ENV variable for this work. So I am not sure why it's the opposite in your case.
Yes, so I used the Dataproc cluster, which had Spark 3.1.2 installed as default. Hence, my command was for the same.
No worries. But I wonder if it's fine if we talk somewhere other than here? I also miss comments here sometimes. Perhaps Slack is good, but I am fine with anything.
I now have questions regarding the dbt + Airflow part. I am very new to SQL, so I really need your help to understand the logic behind the SQL queries you wrote.
In airflow/dags/sql/auth_events.sql, you put a lot of COALESCE. I assume this is to handle the null values. But how about the 9999999999999 for registration? What does it mean?
I think I understand what is going on in the SQL query below. But the SELECT part at the bottom looks kinda random to me. Can you explain the meaning of the string with multiple Ns, two '0's, and two 'NA's?
{{ config(materialized = 'table') }}
SELECT {{ dbt_utils.surrogate_key(['artistId']) }} AS artistKey,
*
FROM (
SELECT
MAX(artist_id) AS artistId,
MAX(artist_latitude) AS latitude,
MAX(artist_longitude) AS longitude,
MAX(artist_location) AS location,
REPLACE(REPLACE(artist_name, '"', ''), '\\', '') AS name
FROM {{ source('staging', 'songs') }}
GROUP BY artist_name
UNION ALL
SELECT 'NNNNNNNNNNNNNNN',
0,
0,
'NA',
'NA'
)
Sure, feel free to reach out over slack. I assume you must be in the Datatalks slack workspace.
For 1 and 2, both are techniques to handle dimensions where the value is null or unknown. It's kinda hard to explain over chat, but the idea is that there should not be any null dimensions in your fact. A null artist represents nothing and also messes up with aggregates and visualizations. Instead, we add a record in the artists dimension to say that all the null artist ids will now be represented by NNNNNNNNNNNNNNN. Just adds a little clarity to your data and helps you track stuff. I hope this makes sense. You'll come across this when you read about dimensional modelling, it's a fairly common topic.
Here's a short explanation on the Kimball group blog about the same.
Activity
ankurchavda commentedon Apr 21, 2022
https://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html#deploying
It is a required dependency for Kafka and Spark Streaming to work.
stephenllh commentedon Apr 21, 2022
I see! Can you explain what the colon symbol (:) means?
ankurchavda commentedon Apr 21, 2022
I guess that is just for separation. It would mean that we want the
spark-sql-kafka-0-10_2.12package for the3.1.2version of Spark. So let's say you are running Spark3.1.1. Then the last value after:will change to3.1.1.stephenllh commentedon Apr 21, 2022
Thanks a lot! I actually have another unrelated question. I have been struggling to find out how EventSim and Kafka work together to create the 4 topics: listen_events, page_view_events, auth_events, status_change_events. Is that hard-coded and only works for the Million Song Dataset? I also saw that each event has different columns, so you assign different Spark schemas for them.
ankurchavda commentedon Apr 21, 2022
Right. Eventsim is currently coded to generate events for a music streaming website. You can change that but that is written in Scala. So if you have experience with that, why not. The topics are also currently hardcoded. You can check the codebase here
stephenllh commentedon Apr 21, 2022
I see. How did you check the columns and the data types of each Kafka event before you define the schemas?
ankurchavda commentedon Apr 21, 2022
From the confluent control center UI. It should be available on port 9021 of your VM. You will have to forward that port to your local machine and open it in the browser.
stephenllh commentedon Apr 21, 2022
I didn't know you can check those out from the control center. Thanks for being helpful.
ankurchavda commentedon Apr 21, 2022
No worries. If you check the youtube video I put out a few days back, you should get an idea of how to navigate through the project. You'll find the link in the README
stephenllh commentedon Apr 26, 2022
Hi. Do you mind answering another question of mine? I hope this is not bothering you.
When I set the environmental variable
KAFKA_ADDRESSto be the external IP, the eventsim somehow couldn't create all 4 topics. The logs suggested that it struggles to find the correct IP address. But when I don't set the environmental variable (and thus setting it to localhost by default as per your code), then it works. Do you have any idea why?ankurchavda commentedon Apr 26, 2022
Hey, no worries, happy to answer your questions.
I am not sure why this would happen. Eventsim is running on the same VM as Kafka, hence it should pick up directly via that. Eventsim doesn't refer to the
KAFKA_ADDRESSvariable as such. Have you mentioned the correct IP, and do you have your9092port open?ankurchavda commentedon Apr 26, 2022
Your Kafka broker will be available at that port for other applications to connect.
stephenllh commentedon Apr 26, 2022
My mistake. It was opened automatically. (Sorry I deleted the comment after I realized that). Now it works. But let me try again after setting the env variable.
stephenllh commentedon Apr 26, 2022
Yeah it doesn't work if I set the
KAFKA_ADDRESSto the external IP of the VM. It works when I unset the env variable.This is what I see in the terminal:
stephenllh commentedon Apr 26, 2022
By the way I just saw this in the Kafka docker compose yaml file
So the KAFKA_ADDRESS env variable does affect stuff.
ankurchavda commentedon Apr 26, 2022
Yes, it does. If you want to connect from an external VM, like the Spark cluster in our case, you need to write to the external IP address instead of the localhost. Hence I added the KAFKA_ADDRESS variable in the docker-compose.yml.
stephenllh commentedon Apr 26, 2022
I just noticed this in the terraform config:
I didn't follow everything in your project so I missed this part. Is this the reason why it doesn't work? But without this step, it does open up port 9092 on my VSCode when I SSH to the Kafka VM though. Or they are totally different things?
stephenllh commentedon Apr 28, 2022
Hey, when I set the port rules, it worked! It turns out this step is important.
ankurchavda commentedon Apr 29, 2022
So on VS Code, you are forwarding the port. It is not the same as opening the port. You are able to forward the port to your local computer because you already authenticated when you did an SSH. On the other hand, you open your port to allow connections from some other VM in the network. By default, no ports are accessible to any other VMs inside or outside the network. You need to specify, which port would you like to open to accept connections.
stephenllh commentedon Apr 30, 2022
Ah I see! This explanation is helpful.
stephenllh commentedon May 2, 2022
I have another question. I see that you set up a Dataprocs server for Spark streaming. Do you think it makes sense if I do everything on a single VM and hence serve it locally? So I was thinking everything would involve "localhost" and the master node would be "local[*]" etc.
ankurchavda commentedon May 3, 2022
You can. However, In the real-world Spark would always be running on some kind of a cluster. So to make things run on the cluster would be a wiser choice if you plan to showcase the infra choices in an interview or somewhere else.
stephenllh commentedon May 3, 2022
I see. Yeah you made a great point.
stephenllh commentedon May 3, 2022
By the way, I am not sure this is a mistake on my behalf, but I think I might have found some potential bugs:
The relevant change I made is the removal of the environmental variable
Otherwise, I couldn't make it work. (Ignore my reply a few days ago, I too thought it worked when I set the env variable to the external IP of the VM, but it actually did not work. My bad.)
I think there is a mistake in the Spark version. In the Spark README, the version 3.0.3 is installed. When I changed it to:
it finally worked.
ankurchavda commentedon May 7, 2022
Hey, apologies for the late reply. There's a lot going on at this point and I kinda missed this comment.
3.1.2installed as default. Hence, my command was for the same.stephenllh commentedon May 7, 2022
No worries. But I wonder if it's fine if we talk somewhere other than here? I also miss comments here sometimes. Perhaps Slack is good, but I am fine with anything.
I now have questions regarding the dbt + Airflow part. I am very new to SQL, so I really need your help to understand the logic behind the SQL queries you wrote.
In
airflow/dags/sql/auth_events.sql, you put a lot ofCOALESCE. I assume this is to handle the null values. But how about the9999999999999forregistration? What does it mean?I think I understand what is going on in the SQL query below. But the
SELECTpart at the bottom looks kinda random to me. Can you explain the meaning of the string with multiple Ns, two '0's, and two 'NA's?ankurchavda commentedon May 7, 2022
Sure, feel free to reach out over slack. I assume you must be in the Datatalks slack workspace.
For 1 and 2, both are techniques to handle dimensions where the value is null or unknown. It's kinda hard to explain over chat, but the idea is that there should not be any null dimensions in your fact. A null artist represents nothing and also messes up with aggregates and visualizations. Instead, we add a record in the artists dimension to say that all the null artist ids will now be represented by
NNNNNNNNNNNNNNN. Just adds a little clarity to your data and helps you track stuff. I hope this makes sense. You'll come across this when you read about dimensional modelling, it's a fairly common topic.Here's a short explanation on the Kimball group blog about the same.