問題タブ [databricks]

Databricks を使用して Spark クラスターを実行しています。curl を使用してサーバーからデータを転送したいと考えています。例えば、

Databricks ノートブック内でこれを行うにはどうすればよいですか (できれば Python ですが、Scala も問題ありません)。

Windows 10 マシンで spark プログラムを実行しています。

以下のsparkプログラムを実行しようとしています

引数を渡してEclipseでアプリケーション実行モードでプログラムを実行すると

なので

src/test/resources/demo.text

以下のエラーで失敗します。

以下は主なエラーメッセージです

入力パスが存在しません: file:/C:/Users/subho/Desktop/code-master/simple-spark-project/src/test/resources/demo.text

以下の場所にファイルがあります。

! ] 1

以下のプログラムを実行すると、成功裏に実行されました。

以下はログファイルです。

よろしくお願いいたします。

Databricks.comの Notebook と Cluster を使用して Spark と Scala を学習するのは初めてです。ファイルを読み込むための非常に単純なコードを次に示します。

しかし、次のようなエラーが発生しました：

java.io.FileNotFoundException: ファイル file:/tmp/myfile.json が存在しません。

org.apache.hadoop.fs.RawLocalFileSystem.getFileStatus(RawLocalFileSystem.java:402) で

org.apache.hadoop.fs.FilterFileSystem.getFileStatus(FilterFileSystem.java:255) で

com.databricks.backend.daemon.dbutils.FSUtils$.cp(DBUtilsCore.scala:82) で

com.databricks.dbutils_v1.impl.DbfsUtilsImpl.cp (DbfsUtilsImpl.scala:40) で

私は Mac を使用しており、ファイルがこの絶対パスに存在することを確認しました。これはスパーク エラーですか? ありがとう！

Spark に非常に大きな DataFrame があり、操作に時間がかかりすぎます。

1,000 万行あります。

より迅速にテストできるようにサンプリングしたいので、次のことを試みています。

これにより、1,000 行しかないデータフレームが保持されると思いました。

しかしredux.count、たとえば、実行にはまだ時間がかかりすぎます (3 分)。

これを 6 GB RAM (DataBricks 製) を搭載した 8 ワーカー ボックスで実行しています。

私は何か間違ったことをしていますか？

ありがとう！

標準に従って定義された既存の JSON スキーマ (ファイル) を使用することは可能ですか?

http://json-schema.org/latest/json-schema-core.html、

JSON データフレームのスキーマを明示的に宣言するには? もしそうなら、例はありますか？JSON イベントを定義する多くの json スキーマ ファイルがありますが、DF がこれらを再利用できるようにするとよいでしょうか?

乾杯

数日前に Databricks を使い始めたばかりで、S3/bucket_name/../raw からいくつかのログ ファイルを取得して処理し、ログに作業したい特定の行が含まれているかどうかを確認し、その行を別のフォルダーに保存しようとしています。 「S3/bucket_name/../processed/」と呼ばれる

これは私がこれまでに試したことです。

必要な行を抽出し、S3 の新しい処理済みフォルダーに新しいファイルを出力できます。しかし、そのファイルにアクセスして結果を出力しようとすると、いくつかのエラーが発生します

エラーメッセージ：

この問題は、出力形式が S3 に保存されていることが原因であると推測しています。ログ ファイルから必要な行を処理して抽出し、その行を別のファイルに保存して S3 に保存し、Databricks の S3 で新しく保存されたファイルを操作する方法はありますか?

Databricks ノートブックを使用してデータセットから作成した RDD があります。

そこから具体的な値を取得しようとすると、シリアル化エラー メッセージが表示されて失敗します。

これが私のデータを取得する場所です（PageCountはCaseクラスです）：

それから私がするとき：

次の例外が発生します。

データセットで同じ試みが機能しますが:

編集 ：

ここに完全なスタックトレースがあります

Kinesis firehose は、ファイル (この場合は時系列の JSON) の永続性を、YYYY/MM/DD/HH (24 の番号付けで時間まで) で分割されたフォルダー階層に管理します...素晴らしい.

Spark 2.0 を使用して、これらのネストされたサブフォルダーを読み取り、すべてのリーフ json ファイルから静的データフレームを作成するにはどうすればよいですか? データフレームリーダーに「オプション」はありますか?

私の次の目標は、これをストリーミング DF にすることです。Firehose によって s3 に保存された新しいファイルは、Spark 2.0 の新しい構造化ストリーミングを使用して、ストリーミング データフレームの一部になります。これはすべて実験的なものであることは承知しています。誰かが以前に S3 をストリーミング ファイル ソースとして使用したことがあり、データが上記のようにフォルダーに分割されていることを願っています。もちろん、Kinesis ストリームのストレートを好むでしょうが、このコネクタには 2.0 の日付がないため、Firehose->S3 が暫定的なものです。

ND: S3 を DBFS にマウントするデータブリックを使用していますが、もちろん EMR や他の Spark プロバイダーでも簡単に使用できます。例を示す共有可能なノートブックがある場合は、ノートブックも参照してください。

乾杯！

SimpleHttpOperatorを使用してREST APIを使用して行った呼び出しを自動化しようとしています。

これは私が行うことができ、うまく機能している呼び出しの例です。

これを SimpleHttpOperator によって生成されるようにフォーマットすると、次のようになり、うまく機能します。

私は今、Airflow によって設定された変数 execution_date にアクセスしようとしています。これは通常、BashOperator または PythonOperator のジンガ テンプレートを使用してアクセスできますが、SimpleHttpOperator ではアクセスできません。

この変数は SimpleHttpOperator では使用できないため、非常に必要です。回避策または SimpleHttpOperator 内でアクセスする方法を知っている場合は、お知らせください。

ありがとう。