使用PySpark和create DataFrame从Bigquery外部表中读取数据
PySpark是一个用于大数据处理的Python库，它提供了与Apache Spark的集成，可以用于分布式数据处理和分析。create DataFrame是PySpark中用于创建数据帧（DataFrame）的方法。数据帧是一种类似于表格的数据结构，可以进行类似于SQL的查询和操作。
BigQuery是Google Cloud提供的一种托管式数据仓库和分析服务，它可以处理大规模的结构化数据。使用PySpark和create DataFrame从BigQuery外部表中读取数据的步骤如下：
- 首先，需要安装PySpark库并配置Spark环境。可以通过pip安装PySpark：
pip install pyspark。 - 导入必要的库和模块：
from pyspark.sql import SparkSession- 创建SparkSession对象：
spark = SparkSession.builder \
.appName("Read from BigQuery") \
.getOrCreate()- 使用SparkSession对象创建DataFrame，并指定BigQuery外部表的相关信息：
project_id = "your-project-id"
dataset_id = "your-dataset-id"
table_id = "your-table-id"
df = spark.read \
.format("bigquery") \
.option("project", project_id) \
.option("dataset", dataset_id) \
.option("table", table_id) \
.load()在上述代码中，需要将"your-project-id"、"your-dataset-id"和"your-table-id"替换为实际的项目ID、数据集ID和表ID。
- 现在，DataFrame df中包含了从BigQuery外部表中读取的数据。可以对其进行各种操作和分析，例如过滤、聚合、排序等。
推荐的腾讯云相关产品：腾讯云数据仓库（TencentDB for TDSQL）、腾讯云分析型数据库（TencentDB for TDSQL Analytics）。
腾讯云数据仓库（TencentDB for TDSQL）是一种高性能、高可用的云数据库产品，适用于大规模数据存储和分析场景。它提供了与Spark等大数据处理框架的集成，可以方便地进行数据导入和查询分析。
腾讯云分析型数据库（TencentDB for TDSQL Analytics）是一种专为大数据分析和数据仓库场景设计的云数据库产品。它提供了高性能的数据导入和查询能力，支持与Spark等大数据处理框架的集成，可以满足复杂的数据分析需求。
更多关于腾讯云数据仓库和腾讯云分析型数据库的详细信息，请访问以下链接：
相关·内容
1回答
我在数据代码中查询bigquery表，
fetch_history_query = 'SELECT * FROM {} WHERE ({}) = ("{}")'.format(bq_table, Columns, filer_value)
print(fetch_history_query)
client = bigquery.Client()
result_history = client.query(fetch_history_query).result().to_dataframe()
print(len(result_history))
上面的代码将re
浏览 4提问于2021-08-10得票数 0
我的用例包括使用Pyspark代码在Bigquery中创建外部表。数据源是Google云存储桶，JSON数据就在这里。我正在将JSON数据读入数据框架，并希望创建一个外部Bigquery表。到目前为止，该表正在创建，但它不是外部表。
df_view.write\
.format("com.google.cloud.spark.bigquery")\
.option('table', 'xyz-abc-abc:xyz_zone.test_table_yyyy')\
.option("temporaryGcsBucke
浏览 3提问于2020-02-26得票数 2
回答已采纳
我正在尝试使用pandas和pyspark从BigQuery中读取数据。我能够获得数据，但不知何故得到以下错误，同时转换为火花DataFrame。
py4j.protocol.Py4JJavaError: An error occurred while calling o28.showString.
: java.lang.IllegalStateException: Could not find TLS ALPN provider; no working netty-tcnative, Conscrypt, or Jetty NPN/ALPN available
at com.goo
浏览 61提问于2019-07-11得票数 0
我正在使用PySpark从Google BigQuery加载数据。
我已经使用以下命令加载了数据：
dfRates = sc.newAPIHadoopRDD(
'com.google.cloud.hadoop.io.bigquery.JsonTextBigQueryInputFormat',
'org.apache.hadoop.io.LongWritable',
'com.google.gson.JsonObject',
conf=conf)
其中conf定义为。
我需要这些数据作为DataFrame，所以我试着
浏览 2提问于2016-03-26得票数 2
我有一个BQ表，它由缺省的_PARTITIONTIME分区。我想把它的一个分区读到Spark dataframe (PySpark)。但是，spark.read应用编程接口似乎不能识别分区列。下面是代码(不起作用)：
table = 'myProject.myDataset.table'
df = spark.read.format('bigquery').option('table', table).load()
df_pt = df.filter("_PARTITIONTIME = TIMESTAMP('2019-01-3
浏览 0提问于2019-07-31得票数 1
我正在尝试从外部表中读取数据，外部表是使用pyspark笔记本配置在突触无服务器池中的(我希望它在一个dataframe中)，但是我无法做到这一点。有人能指点我怎么做吗。我尝试了下面的代码。
server_name = "jdbc:sqlserver://<<sservername>>-ondemand.sql.azuresynapse.net"
database_name = "<<dbname>>"
url = server_name + ";" + "databaseName="
浏览 4提问于2022-04-20得票数 0
1回答
我有大约500-700 csv文件的日志加载每周。文件名是每周随机生成的。这些文件存储在google云存储上，pyspark在google dataproc上运行，有几个工作节点。
我已经创建了一个pyspark文件，它基本上只是迭代文件并按原样加载到bigquery中。我以为我会在bigquery表上看到增量加载，但似乎pyspark会将整个gz输入文件合并到dataframe中，然后立即处理它。所以我在大约90分钟后得到了结果。
我如何编写pyspark代码，以便每次读取文件时，都会将其写入bigquery。总体时间可能不会有太大变化，但至少我如何才能看到部分写入bigquery的数据？
浏览 22提问于2021-10-10得票数 1
1回答
我是PySpark新手，正在尝试读取HDFS文件(上面创建了hive表)并创建PySpark数据帧。通过PySpark读取配置单元表非常耗时。有没有什么方法可以动态获取hive列名(用作dataframe中的模式)？ 我希望将文件位置、表名和数据库名作为输入传递给aa程序/函数，以便从配置单元元数据(可能是元数据xml)中获取模式/列名称，并作为dataframe返回。 请指教
浏览 15提问于2019-03-06得票数 0
回答已采纳
1回答
我们正在尝试将pyspark脚本从一个前提迁移到GCP平台，该脚本通过数据转换在Hive中创建和删除表。
蜂巢被BigQuery所取代。在本例中，使用连接器将单元读和写转换为bigquery读写。
然而，问题在于通过spark创建和删除bigquery表，因为spark将默认在由蜂窝转移支持的蜂巢上运行创建和删除查询，而不是在大型查询上运行。
我想检查一下是否有计划将DDL语句支持以及星火-bigquery-连接器的一部分结合在一起。
此外，从体系结构的角度看，是否有可能将spark的转移建立在bigquery上，以便任何create或drop语句都可以从spark运行在bigquery上。
浏览 7提问于2022-03-30得票数 2
回答已采纳
从正式文档中可以看到，它首先将表加载到Spark中，然后使用.sql()执行查询。
words = spark.read.format('bigquery') \
.option('table', 'bigquery-public-data:samples.shakespeare') \
.load()
words.createOrReplaceTempView('words')
# Perform word count.
word_count = spark.sql(
'SELECT word, SUM
浏览 3提问于2019-11-19得票数 1
回答已采纳
我希望使用PySpark将数据保存到一个Hive表中。
指出：
"spark.sql.hive.convertMetastoreParquet:当设置为false时，Spark将使用Hive SerDe来处理拼花表，而不是支持内置的。“
查看，似乎可以设置此属性：
from pyspark.sql import HiveContext
sqlContext = HiveContext(sc)
sqlContext.sql("SET spark.sql.hive.convertMetastoreParquet=false")
# code to create
浏览 2提问于2015-07-17得票数 9
回答已采纳
我正在处理datalab，但是当我试图查询bigquery中的表时，会得到以下错误：
异常:无效:读取表时出错：.错误消息:读取电子表格失败。错误:没有找到带有Google作用域的OAuth令牌。
这种情况只发生在与google驱动器表链接的表中。
现在，在gcp中启用google驱动器应用程序。
from google.cloud import bigquery
client = bigquery.Client()
sql = """
SELECT * FROM `proyect-xxxx.set_xxx.table_x` LIMIT 1000
"&
浏览 1提问于2019-04-19得票数 0
我是dataproc集群和PySpark的新手，因此，在寻找代码以将表从bigquery加载到集群的过程中，我遇到了下面的代码，并且无法弄清楚我应该为这段代码中的用例修改什么，以及我们在输入目录中提供了什么作为输入
from pyspark.context import SparkContext
from pyspark.sql.session import SparkSession
import subprocess
sc = SparkContext()
spark = SparkSession(sc)
bucket = spark._jsc.hadoopConfiguration
浏览 1提问于2020-05-31得票数 1
1回答
我从Bigquery中的一个视图(在配置spark会话之后导入该视图)中获得了pyspark中的数据：
config = pyspark.SparkConf().setAll([('spark.executor.memory', '10g'),('spark.driver.memory', '30G'),\
('spark.jars.packages', 'com.google.cloud.spark:spark-bigquery-with-
浏览 2提问于2021-01-06得票数 1
1回答
我使用BigQuery存储“原生”BigQuery表中的数据，并查询存储在中的数据。根据文档，可以使用两种类型的表查询外部源:永久表和临时外部表。
考虑以下场景:每天都有一些拼花文件是用GCS编写的，并且在一定频率下，我希望在存储在BigQuery表中的数据和存储在拼花文件中的数据之间进行连接。如果我创建一个永久外部表，然后更新下面的文件，表的内容是否也会自动更新，还是必须从新文件中重新创建它？这种情况的最佳做法是什么？
浏览 11提问于2020-08-25得票数 2
回答已采纳
3回答
我正在尝试使用官方的python客户机库将一个相对较大的pandas dataframe df加载到Google BigQuery表table_ref中。
到目前为止，我已经尝试了两种不同的方法：
1)直接从内存中的数据中加载表
client = bigquery.Client()
client.load_table_from_dataframe(df, table_ref)
2)将数据存储保存到位于uri parquet_uri的Google中的一个拼花文件中，并从该文件加载表：
df.to_parquet(parquet_uri)
client = bigquery.Client()
cl
浏览 12提问于2020-03-29得票数 5
回答已采纳
我使用bigquery客户端对象将一些CSV文件(位于云存储中)上载到bigquery表中。
我成功地将数据上载到bigquery表中，但我希望将目标表更改为已分区表。分区将是文件名中的日期。
文件名是CSV文件中的一个列，与CSV文件名相同。
这就是我如何从文件名(假设文本是文件名)中提取数据的方法，date1稍后将用作我们的分区：
text = 'sales_2022-09-09T21-27-05_018787'
match = re.search(r'\d{4}-\d{2}-\d{2}', text)
date1 = datetime.strptime(m
浏览 7提问于2022-11-21得票数 0
我有一个数据流代码，它从gs://中的存储桶读取CSV文件，并将该CSV文件摄取到BigQuery表中。BigQuery表已经创建。下面的代码工作正常。
class DataIngestion:
"""A helper class which contains the logic to translate the source csv file into a format BigQuery will accept."""
def parse_method(self, string_input):
values
浏览 12提问于2022-02-03得票数 0
回答已采纳
我正在把一个表格从BigQuery读到Spark。该表具有以下列: id(int)、string1(字符串)、string2(字符串)、string3(字符串)、date(日期)、timestamp(int)。现在，我想将结果数据帧中的行根据id写到BigQuery中的不同表中。例如，有700行id为1的行，我想创建一个表project.dataset.id1，并将这700行(所有列)都写到该表中，然后所有id为2的行都应该写到一个表project.dataset.id2中，依此类推。
如果我正在写文本文件，我将使用write.partitionBy('id')，当我写到Big
浏览 1提问于2019-11-19得票数 0
我尝试从bigquery中读取一个表，然后将其转换为dataframe，最后删除该表。但是由于数据帧是延迟计算的，所以会发生什么呢?它首先删除表，然后尝试读取它。在将表加载到dataframe之后，有什么方法可以删除它吗？
浏览 3提问于2017-01-18得票数 0
如何在Azure数据湖存储上创建外部Delta表？我目前正在从事一个迁移项目(从Pyspark/Hadoop到Azure)。我找不到很多关于在Azure中创建非托管表的文档。下面是我目前能够在Pyspark/Hive/HDFS设置中执行的一系列操作，不知道如何在Azure上建立相同的操作。
按顺序行动-
如果存在，则创建一个dataframe DF DF.write.insertInto("table")create 拖放单元外部表，使用DF1.write.insertInto("table")将dataframe DF加载到此外部表--如果存在，则将dataf
浏览 3提问于2022-03-30得票数 0
最近，我遇到了一个问题，当我使用pyspark将数据写入BigQuery中时。这里是：
pyspark.sql.utils.IllegalArgumentException:必须通知U‘informed或持久性GCS桶
在对问题进行研究后，我发现暂时性的GCS桶要提到spark.conf。
bucket = "temp_bucket"
spark.conf.set('temporaryGcsBucket', bucket)
我认为在Biquery中为像Hive这样的表创建一个文件是没有概念的。
我想知道更多关于它的情况，为什么我们需要临时gcs将数据写入
浏览 6提问于2021-06-11得票数 1
回答已采纳
2回答
我们有从多个源表读取数据并根据业务规则连接和应用映射的场景。在某些情况下，从几个表中读取的数据可以用于多个目标加载。那么，为了避免在运行不同的模块时多次读取相同的数据，有什么选择可以在不同的pyspark模块中使用相同的dataframe输出吗？
df1 = spark.sql(select * from table1)
df2 = spark.sql(select * from table2)
df_out = df1.join(df2, ['customer_id'], inner)
我想在pyspark_module1.py和pyspark_module2.py中使用d
浏览 0提问于2020-11-07得票数 0
我在Windows 10上使用Spark 2.4和Ignite 2.8 当我把pyspark dataframe放到点火表中时，它保存得很好。 但是，当我将这些数据从Ignite读回Pyspark时，dataframe又回到了系统默认编码(CP1251)。我不明白如何用正确的编码重新获得正确的数据。 以下是示例代码： import pandas as pd
import pyspark
import os
# Starting local SPARK
os.environ['PYSPARK_SUBMIT_ARGS'] = " --driver-memory 18g
浏览 13提问于2020-04-09得票数 0
回答已采纳
从Google控制台，我们可以手动创建外部bigquery表，该表将链接到Google云存储Json新行分隔文件。
但是在bigquery客户端库中，我找不到示例代码允许我创建bigquery外部表链接到云存储保存的json文件。
function create_table($projectId, $datasetId, $tableId, $schema)
{
$bigQuery = new BigQueryClient([
'projectId' => $projectId,
]);
$dataset = $bigQuery->dataset
浏览 7提问于2017-03-23得票数 0
回答已采纳
3回答
我有以下用例:需要使用PySpark分析大量结构化数据。数据当前为CSV格式。我正在尝试找出在谷歌云上存储数据的最佳方式。我的理解是HDFS不会工作，因为每次集群关闭，HDFS数据都会消失，所以我每次都必须从CSV转换到HDFS，这很耗时。似乎正确的策略是使用BigQuery，但是我不能确定BigQuery是否是持久性的？
浏览 20提问于2017-06-28得票数 0
回答已采纳
尝试从bigquery读取数据到jupyter笔记本，并使用吡火花库。apache和java的所有文件都被下载到我的C:驱动器中。阅读和观看教程视频，但没有一个似乎是有效的。寻求指导
代码：
import pyspark
import findspark
from pyspark import SparkContext,SparkConf
from pyspark.sql import SparkSession
from pyspark.sql.functions import window, col, year, month, aggregate, date_add,
timestam
浏览 9提问于2022-02-04得票数 -1
我的pyspark数据帧中有多语言字符。在将数据写入bigquery之后，它向我显示了奇怪的字符，因为它的默认编码方案(utf-8)。如何使用pyspark / dataproc将Bigquery中的编码更改为ISO_8859_1？
浏览 0提问于2020-05-06得票数 1
我正在尝试从用户管理的朱庇特笔记本实例中读取一些BigQuery数据(ID：my-project.mydatabase.mytable原始名称受保护)，在工作台中。我尝试的是中的灵感，更具体地说，代码是(请阅读一些关于代码本身的附加注释)：
from pyspark.sql import SparkSession
from pyspark.sql.functions import udf, col
from pyspark.sql.types import IntegerType, ArrayType, StringType
from google.cloud import bigquery
浏览 15提问于2022-08-09得票数 1
回答已采纳
根据Dataproc ，它具有“与BigQuery的本地和自动集成”。
我在BigQuery有一张桌子。我希望阅读该表，并使用我创建的Dataproc集群(使用PySpark作业)对其执行一些分析。然后将此分析的结果写回BigQuery。您可能会问：“为什么不直接用BigQuery进行分析呢？”--原因是我们正在创建复杂的统计模型，而SQL太高，无法开发它们。我们需要类似Python或R的东西，因此需要Dataproc。
它们有任何Dataproc + BigQuery示例可用吗？我什么都找不到。
浏览 3提问于2015-10-06得票数 11
回答已采纳
我将一个df从pyspark导出到BigQuery。df包含包含数组元素的列，如何将数组转换为连接字符串？每当我尝试查询导出的BigQuery表的数组列时，都会得到以下错误。
Error: Cannot access field element on a value with type ARRAY<STRUCT<element STRING>>
下面是导出到BigQuery的pyspark中的df：
+-----------+----------+-------------------+------------------+
| antecedent|conseque
浏览 48提问于2020-06-03得票数 2
1回答
我在Google Cloud中使用BigQuery和Dataproc。两者都在同一个项目中，我们称它为" project -123“。我使用Composer (Airflow)来运行代码。
我有一个简单的python脚本test_script.py，它使用pyspark从bigquery公共数据集中的一个表中获取读取数据：
if __name__ == "__main__":
# Create Spark Cluster
try:
spark = SparkSession.builder.appName("test_script&
浏览 19提问于2020-03-06得票数 1
我正在尝试使用PySpark读取表格形式的BigQuery。 我尝试过以下几种方法 table = 'my-project-id.project-dataset.test_table_spark'
df = spark.read.format('bigquery').option('table', table).load() 然而，我得到了这个错误 : java.lang.ClassNotFoundException: Failed to find data source: bigquery. Please find packages at h
浏览 19提问于2019-10-01得票数 2
回答已采纳
我试图限制用户在BigQuery中只能看到特定的数据集。我希望用户只能运行查询和创建表。因此，我共享了dataset权限，并为他们提供了角色BigQuery Data Viewer (查看数据)和BigQuery User (运行查询和创建表)。但每当我运行查询时，我都会得到以下错误：
访问被拒绝:项目xxx:用户在项目xxx中没有bigquery.jobs.create权限。
我检查了和BigQuery用户角色是否应该具有bigquery.jobs.create权限。我遗漏了什么？
浏览 2提问于2020-03-02得票数 0
根据BigQuery的python文档(Version3.3.2)，有一个用于insert_rows_from_dataframe (dataframe: pandas.DataFrame)的方法，但是PyArrow没有类似的方法。
insert_rows_from_dataframe(
table: Union[
google.cloud.bigquery.table.Table,
google.cloud.bigquery.table.TableReference,
str,
],
dataframe,
sel
浏览 12提问于2022-09-19得票数 0
上下文 我正在尝试使用Composer，DataProc和BigQuery在谷歌云平台上建立一个摄取管道。我在BigQuery中有一个表，其中包含数据源及其相关文件的记录。因此，如果我需要摄取5个文件，那么在BigQuery表中就有5条记录。明天可能会有不同数量的文件。因此，我考虑在我的DAG中动态构建任务。 概要设计如下： 执行一个函数，以Pandas dataframe (或dict，两者都可以)的形式从BigQuery获取数据。 迭代数据帧 对于数据帧中的每一行，创建一个DataProcSparkOperator，其中包含有关文件和相应参数的详细信息 此设置运行良好。我可以在Airfl
浏览 51提问于2021-02-25得票数 0
回答已采纳
我想请你帮个忙。我自己已经为这个任务解决了很多问题，但不幸的是，我不能解决下面的问题。我需要使用Spark (即PySpark)从谷歌BigQuery下载一个公共数据集(例如，BigQuery- public -data:samples.shakepeare)到我的PC上。对我来说，它应该是类似于"HelloWorld“的例子。我已经在Ubuntu6.0.10( VirtualBox 18.04LTS)中本地安装了Spark。首先，我想从上面提到的表格中展示一些数据作为开始。我为此创建了以下代码(代码位于名为spark_download_data_from_bq.py的文件中)。
#!
浏览 0提问于2020-03-03得票数 1
我正在创建一个到Hive的连接字符串，并在该连接上的Hive表上运行一些SELECT查询。
在对检索到的数据执行一些转换之后，我将创建一个数据框架df_student_credits，如下所示
NAME_STUDENT_INITIAL CREDITS_INITIAL NAME_STUDENT_FINAL CREDITS_FINAL LOAD_DATE
John 23 John 25 21/03/2017
Alan
浏览 0提问于2019-08-21得票数 2
回答已采纳
我有如下数据集：
我正在使用PySpark解析数据，然后使用下面的代码创建一个DataFrame：
from pyspark.sql import SparkSession
from pyspark.sql import Row
from pyspark.sql import functions as f
def parseInput(line):
fields = line.split(',')
stationID=fields[0]
entryType=fields[2]
temperature= fields[3]*0.3
浏览 2提问于2020-07-11得票数 0
回答已采纳
我正在尝试使用spark sql从pyspark dataframe创建一个hive paritioned表。下面是我正在执行的命令，但得到了一个错误。错误消息如下。 df.createOrReplaceTempView(df_view)
spark.sql("create table if not exists tablename PARTITION (date) AS select * from df_view") 错误：pyspark.sql.utils.ParseException:u"\nmismatched input 'PARTITION'
浏览 113提问于2021-09-19得票数 0
回答已采纳
我尝试将数据帧存储到Spark 1.3.0 (PySpark)中的持久化蜂窝表。这是我的代码：
sc = SparkContext(appName="HiveTest")
hc = HiveContext(sc)
peopleRDD = sc.parallelize(['{"name":"Yin","age":30}'])
peopleDF = hc.jsonRDD(peopleRDD)
peopleDF.printSchema()
#root
# |-- age: long (nullable = true)
#
浏览 0提问于2015-05-14得票数 9
回答已采纳
我正在Databricks notebook上编写pyspark脚本来插入/更新/查询cassandra表，但是我找不到从表中删除行的方法，我尝试了spark sql： spark.sql("DELETE from users_by_email where email_address IN ('abc@test.com')") 我也不认为使用dataframe删除数据是可能的。有什么变通方法吗？
浏览 25提问于2019-12-17得票数 0
回答已采纳
我正在尝试将Data报告连接到表和BigQuery的视图。在BigQuery中，表从位于receives中的GoogleSheet接收数据，视图正在查询这个表，但添加了一些新字段。当我试图将我的图表连接到这些数据源之一时，我会得到以下错误：
BigQuery错误:访问被拒绝: BigQuery BigQuery:在获得驱动器凭据时被拒绝的权限。
原因是什么？我该怎么做才能解决这个问题？
我建议我是纸张，BigQuery项目和DS报告的所有者。
谢谢
浏览 0提问于2021-04-01得票数 3
我在用数据库。假设我有两个(我使用的是PySpark)：
df_source
df_target
如果df_source有以下模式：
root
|-- name: string (nullable = false)
|-- id: long (nullable = false)
|-- age: long (nullable = true)
df_target有以下模式：
root
|-- name: string (nullable = true)
|-- id: long (nullable = false)
|-- age: long (nullable =
浏览 1提问于2021-12-27得票数 0
回答已采纳
我在运行一个由卡夫卡，斯派克和卡桑德拉组成的单节点集群。都在同一台机器上。
从一个简单的Python脚本中，我每5秒将一些虚拟数据流到Kafka主题中。然后，使用Spark结构化流，我将这个数据流(每次一行)读入PySpark DataFrame中，其中包含startingOffset = latest。最后，我试图将这一行附加到已经存在的Cassandra表中。
我一直在跟踪()和()。
有一行数据正在成功地写入Cassandra表中，但我的问题是，每次覆盖时，都会被覆盖，而不是将追加到表的末尾。我可能做错什么了？
这是我的密码：
用于创建kafkaspark密钥空间的CQL，后面是卡桑德拉
浏览 1提问于2020-10-21得票数 0
回答已采纳
如何在DataFrame中为流PySpark设置架构。
from pyspark.sql import SparkSession
from pyspark.sql.functions import explode
from pyspark.sql.functions import split
# Import data types
from pyspark.sql.types import *
spark = SparkSession\
.builder\
.appName("StructuredNetworkWordCount")\
.getOrCr
浏览 3提问于2016-12-29得票数 3
回答已采纳
我们在BigQuery表中有大量的原始数据(100亿到4000亿)。我们需要处理这些数据，以转换和创建星型模式表的形式(bigquery中可能是不同的数据集)，然后可以通过atscale访问这些数据。
需要以下两种选择之间的利弊：
1.在BigQuery中写入复杂的SQL，读取数据源数据集，然后加载到目标数据集(由Atscale使用)。
2.将PySpark或MapReduce与来自Dataproc的BigQuery连接器一起使用，然后将数据加载到BigQuery目标数据集中。
转换的复杂性包括在不同粒度上连接多个表，使用分析函数获取所需的信息等。
目前，该逻辑是在vertica中实现的，使用
浏览 10提问于2020-01-14得票数 0
2回答
我正在尝试将运行在私有数据中心上的火花作业与BigQuery连接起来。我创建了服务帐户，获得了私有JSON密钥，并获得了对我想要查询的数据集的读取访问权。但是，当我尝试与星火集成时，我正在接收User does not have bigquery.tables.create permission for dataset xxx:yyy.。我们是否需要创建表权限来使用BigQuery从表中读取数据？
下面是控制台上打印的响应，
{
"code" : 403,
"errors" : [ {
"domain" : "global
浏览 7提问于2020-12-09得票数 0
TL;DR: asyncio vs multi-processing vs threading vs some other solution，将读取GCS文件的for循环并行化，然后将这些数据附加到熊猫数据中，然后写入BigQuery.
我想做一个并行的python函数，它从GCS目录中读取数十万个小.json文件，然后将这些.jsons转换成熊猫数据格式，然后将熊猫数据写入BigQuery表。
下面是该函数的一个非并行版本：
import gcsfs
import pandas as pd
from my.helpers import get_gcs_file_list
def load_g
浏览 1提问于2020-07-23得票数 3
我已经在GCS中创建了一个NEWLINE DELIM JSON文件。我还在同一个JSON文件上创建了一个外部表，并且能够从BigQuery UI读取数据。 我想使用PySpark访问外部表数据并创建一个dataframe，然后从Dataproc运行相同的作业。下面是我写的代码片段： #!/usr/bin/python
import sys
import json
from pyspark.sql.functions import udf, lit, when, date_sub
from pyspark.sql.types import ArrayType, IntegerType, Str
浏览 23提问于2020-08-14得票数 2
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
