python-3.x - 在 Colab.Research.Google 上使用 Python 从位于 Google 云端硬盘中的文件创建 Spark RDD

Question

我在 Google 的 Colab.Research 平台上成功运行了 Python 3/Spark 2.2.1 程序:

!apt-get update
!apt-get install openjdk-8-jdk-headless -qq > /dev/null
!wget -q http://apache.osuosl.org/spark/spark-2.2.1/spark-2.2.1-bin-hadoop2.7.tgz
!tar xf spark-2.2.1-bin-hadoop2.7.tgz
!pip install -q findspark

import os
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
os.environ["SPARK_HOME"] = "/content/spark-2.2.1-bin-hadoop2.7"

import findspark
findspark.init()
from pyspark.sql import SparkSession
spark = SparkSession.builder.master("local[*]").getOrCreate()

当我使用以下命令将文本文件从本地计算机上传到 Unix VM 时，这非常有效

from google.colab import files
datafile = files.upload()

然后按如下方式阅读它们:

textRDD = spark.read.text('hobbit.txt').rdd

到目前为止一切顺利..

当我尝试读取位于我的 Google Drive colab 目录中的文件时，我的问题就开始了。

按照说明我已经对用户进行了身份验证并创建了一个驱动器服务

from google.colab import auth
auth.authenticate_user()

from googleapiclient.discovery import build
drive_service = build('drive', 'v3')

之后我就可以访问位于驱动器中的文件，如下所示:

file_id = '1RELUMtExjMTSfoWF765Hr8JwNCSL7AgH'

import io
from googleapiclient.http import MediaIoBaseDownload

request = drive_service.files().get_media(fileId=file_id)
downloaded = io.BytesIO()
downloader = MediaIoBaseDownload(downloaded, request)
done = False
while done is False:
  # _ is a placeholder for a progress object that we ignore.
  # (Our file is small, so we skip reporting progress.)
  _, done = downloader.next_chunk()

downloaded.seek(0)
print('Downloaded file contents are: {}'.format(downloaded.read()))

Downloaded file contents are: b'The king beneath the mountain\r\nThe king of ......

即使这样也很完美..

downloaded.seek(0)
print(downloaded.read().decode('utf-8'))

获取数据

The king beneath the mountain
The king of carven stone
The lord of silver fountain ...

最终出错的地方是我尝试获取这些数据并将其放入 spark RDD 的地方

downloaded.seek(0)
tRDD = spark.read.text(downloaded.read().decode('utf-8'))

我得到了错误..

AnalysisException: 'Path does not exist: file:/content/The king beneath the mountain\ ....

显然，我没有使用正确的方法/参数将文件读入 spark。我已经尝试了很多描述的方法

如果有人能帮我弄清楚如何读取这个文件以进行后续处理，我将不胜感激。

Answer 1

此问题的完整解决方案可在另一个可用的 StackOverflow 问题中找到 at this URL .

这是 notebook演示此解决方案的位置。

我已经测试过了，它有效!