$ docker-compose down $ docker rmi -f docker_ckan docker_db $ docker rmi $(docker images -f dangling=true -q) $ docker volume rm docker_ckan_config docker_ckan_home docker_ckan_storage docker_pg_data docker_solr_data $ docker-compose build $ docker-compose up -d $ docker-compose restart ckan # give the db service time to initialize the db cluster on first run
dobachi@ubu18:~$ sudo stat /sys/fs/cgroup/memory/memsw.limit_in_bytes stat: '/sys/fs/cgroup/memory/memsw.limit_in_bytes' を stat できません: そのようなファイルやディレクトリはありません
$ minikube kubectl -- logs pi-43bff77e450bdba3-driver (snip) 22/01/10 17:32:02 INFO TaskSchedulerImpl: Killing all running tasks in stage 0: Stage finished 22/01/10 17:32:02 INFO DAGScheduler: Job 0 finished: reduce at SparkPi.scala:38, took 0.824388 s Pi is roughly 3.142475712378562 22/01/10 17:32:02 INFO SparkUI: Stopped Spark web UI at http://pi-43bff77e450bdba3-driver-svc.default.svc:4040 22/01/10 17:32:02 INFO KubernetesClusterSchedulerBackend: Shutting down all executors (snip)
# The format version of this config file version:1 # Config shares/schemas/tables to share shares: -name:"share1" schemas: -name:"schema1" tables: -name:"table1" location:"s3a://test_bucket/delta_table" # Set the host name that the server will use host:"localhost" # Set the port that the server will listen on port:18080 # Set the url prefix for the REST APIs endpoint:"/delta-sharing" # Set the timeout of S3 presigned url in seconds preSignedUrlTimeoutSeconds:900 # How many tables to cache in the server deltaTableCacheSize:10 # Whether we can accept working with a stale version of the table. This is useful when sharing # static tables that will never be changed. stalenessAcceptable:false # Whether to evaluate user provided `predicateHints` evaluatePredicateHints:false
Caused by: org.apache.hadoop.fs.s3a.AWSS3IOException: doesBucketExist on test: com.amazonaws.services.s3.model.AmazonS3Exception: Bad Request (Service: Amazon S3; Status Code: 400; Error Code: 400 Bad Request; Request ID: null; S3 Extended Request ID: null), S3 Extended Request ID: null: Bad Request (Service: Amazon S3; Status Code: 400; Error Code: 400 Bad Request; Request ID: null; S3 Extended Request ID: null) at org.apache.hadoop.fs.s3a.S3AUtils.translateException(S3AUtils.java:194) at org.apache.hadoop.fs.s3a.S3AFileSystem.verifyBucketExists(S3AFileSystem.java:335) at org.apache.hadoop.fs.s3a.S3AFileSystem.initialize(S3AFileSystem.java:280) at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:3247) at org.apache.hadoop.fs.FileSystem.access$200(FileSystem.java:121) at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:3296) at org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem.java:3264) at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:475) at org.apache.hadoop.fs.Path.getFileSystem(Path.java:356) at io.delta.standalone.internal.DeltaSharedTable.$anonfun$deltaLog$1(DeltaSharedTableLoader.scala:76) at io.delta.standalone.internal.DeltaSharedTable.withClassLoader(DeltaSharedTableLoader.scala:95) at io.delta.standalone.internal.DeltaSharedTable.<init>(DeltaSharedTableLoader.scala:74) at io.delta.standalone.internal.DeltaSharedTableLoader.$anonfun$loadTable$1(DeltaSharedTableLoader.scala:53) at com.google.common.cache.LocalCache$LocalManualCache$1.load(LocalCache.java:4693) at com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3445) at com.google.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2194) ... 60 more
# The format version of this config file version:1 # Config shares/schemas/tables to share shares: -name:"share1" schemas: -name:"schema1" tables: -name:"table1" location:"s3a://<your configuration>" # Set the host name that the server will use host:"localhost" # Set the port that the server will listen on port:80 # Set the url prefix for the REST APIs endpoint:"/delta-sharing" # Set the timeout of S3 presigned url in seconds preSignedUrlTimeoutSeconds:900 # How many tables to cache in the server deltaTableCacheSize:10 # Whether we can accept working with a stale version of the table. This is useful when sharing # static tables that will never be changed. stalenessAcceptable:false # Whether to evaluate user provided `predicateHints` evaluatePredicateHints:false
defload_as_pandas(url: str) -> pd.DataFrame: """ Load the shared table using the give url as a pandas DataFrame. :param url: a url under the format "<profile>#<share>.<schema>.<table>" :return: A pandas DataFrame representing the shared table. """ profile_json, share, schema, table = _parse_url(url) profile = DeltaSharingProfile.read_from_file(profile_json) return DeltaSharingReader( table=Table(name=table, share=share, schema=schema), rest_client=DataSharingRestClient(profile), ).to_pandas()
DATA COLLABORATION
GUIDE にデータコラボレーションに関するまとめが記載されている。
データコラボレーションの具体例（パートナーとアライアンス、チャネル販売、など）に触れながら、
データエスクローについて紹介している。
What
is data
collaboration?にデータコラボレーションによってもたらされるメリットなどに少し触れている生地記事。
「データブリッジ」というクエリを移動させる方式（考え方？）が紹介されていた。
データコラボレーションは、線シティブナデータを共有することなく、サイロ化したデータを組み合わせられる仕組みであると紹介されている。
```scala def main(args: Array[String]): Unit = { val ns = parser.parseArgsOrFail(args) val serverConfigPath = ns.getString("config") val serverConf = ServerConfig.load(serverConfigPath) start(serverConf).blockUntilShutdown() }
コンフィグのロード
渡された設定ファイルのPATHを用いて、設定を読み込む。
1 2 3 4 5 6 7 8 9 10 11 12 13 14
/** * Load the configurations for the server from the config file. If the file name ends with * `.yaml` or `.yml`, load it using the YAML parser. Otherwise, throw an error. */ defload(configFile: String): ServerConfig = { if (configFile.endsWith(".yaml") || configFile.endsWith(".yml")) { val serverConfig = createYamlObjectMapper.readValue(newFile(configFile), classOf[ServerConfig]) serverConfig.checkConfig() serverConfig } else { thrownewIOException("The server config file must be a yml or yaml file") } }
defstart(serverConfig: ServerConfig): Server = { lazyval server = { updateDefaultJsonPrinterForScalaPbConverterUtil() val builder = Server.builder() .defaultHostname(serverConfig.getHost) .disableDateHeader() .disableServerHeader() .annotatedService(serverConfig.endpoint, newDeltaSharingService(serverConfig): Any) if (serverConfig.ssl == null) { builder.http(serverConfig.getPort) } else { builder.https(serverConfig.getPort) if (serverConfig.ssl.selfSigned) { builder.tlsSelfSigned() } else { if (serverConfig.ssl.certificatePasswordFile == null) { builder.tls( newFile(serverConfig.ssl.certificateFile), newFile(serverConfig.ssl.certificateKeyFile)) } else { builder.tls( newFile(serverConfig.ssl.certificateFile), newFile(serverConfig.ssl.certificateKeyFile), FileUtils.readFileToString(newFile(serverConfig.ssl.certificatePasswordFile), UTF_8) ) } } } if (serverConfig.getAuthorization != null) { // Authorization is set. Set up the authorization using the token in the server config. val authServiceBuilder = AuthService.builder.addOAuth2((_: ServiceRequestContext, token: OAuth2Token) => { // Use `MessageDigest.isEqual` to do a time-constant comparison to avoid timing attacks val authorized = MessageDigest.isEqual( token.accessToken.getBytes(UTF_8), serverConfig.getAuthorization.getBearerToken.getBytes(UTF_8)) CompletableFuture.completedFuture(authorized) }) builder.decorator(authServiceBuilder.newDecorator) } builder.build() } server.start().get() server }
一番最後の箇所の通り、
1 2
server.start().get() server
server は、 Armeria
のビルダを用いてインスタンス化されたサーバを起動する。
なお、startメソッド内ではTLS周りの設定、トークンの設定などが行われる。
なお、サーバに渡されるクラスは以下の通り、
1 2 3 4 5
val builder = Server.builder() .defaultHostname(serverConfig.getHost) .disableDateHeader() .disableServerHeader() .annotatedService(serverConfig.endpoint, newDeltaSharingService(serverConfig): Any)
/** Return the current table version */ deftableVersion: Long = withClassLoader { val snapshot = deltaLog.snapshot validateDeltaTable(snapshot) snapshot.version }
/** * Run `func` under the classloader of `DeltaSharedTable`. We cannot use the classloader set by * Armeria as Hadoop needs to search the classpath to find its classes. */ privatedefwithClassLoader[T](func: => T): T = { val classLoader = Thread.currentThread().getContextClassLoader if (classLoader == null) { Thread.currentThread().setContextClassLoader(this.getClass.getClassLoader) try func finally { Thread.currentThread().setContextClassLoader(null) } } else { func } }
